bec/csaxs_bec
0
0
Fork 1
Code Issues 15 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

feat: first draft for jungfraujoch ophyd

Browse Source
This commit is contained in:
gac-x12sa
2024-10-09 16:38:25 +02:00
parent 9dd96c9f6e
commit 39491e3189
5 changed files with 288 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
csaxs_bec/device_configs/ddg_template.yaml
View File

@@ -25,6 +25,17 @@ ddg:
enabled: true
readoutPriority: async
softwareTrigger: True
eiger_jfjoch:
description: DelayGenerator for detector triggering
deviceClass: csaxs_bec.devices.jungfraujoch.eiger_jungfrau_joch.Eiger9McSAXS
deviceConfig:
deviceTags:
- cSAXS
- eiger9m
onFailure: buffer
enabled: true
readoutPriority: async
softwareTrigger: False
simulated_monitor:
readoutPriority: monitored
deviceClass: ophyd_devices.SimMonitor

33
csaxs_bec/devices/epics/delay_generator_csaxs.py
View File

@@ -207,10 +207,15 @@ class DDGSetup(DDGCustomMixin):
# scantype "jjf_test"
scan_name = self.parent.scaninfo.scan_msg.content["info"].get("scan_name", "")
if scan_name == "jjf_test":
exp_time = self.parent.scaninfo.exp_time
readout = self.parent.scaninfo.readout_time
num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
# exp_time = self.parent.scaninfo.exp_time
# readout = self.parent.scaninfo.readout_time
# num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
# total_exposure = exp_time+readout
exp_time = 480e-6#self.parent.scaninfo.exp_time
readout = 20e-6#self.parent.scaninfo.readout_time
total_exposure = exp_time+readout
num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
num_burst_cycle = int(num_burst_cycle * self.parent.scaninfo.exp_time/total_exposure)
delay = 0
delay_burst = self.parent.delay_burst.get()
@@ -219,6 +224,20 @@ class DDGSetup(DDGCustomMixin):
self.parent.set_channels(signal='width', value=exp_time)
self.parent.set_channels(signal='delay', value=delay)
self.parent.burst_enable(count=num_burst_cycle, delay=delay_burst, period=total_exposure, config="first")
logger.info(f"{self.parent.name}: On stage with n_burst: {num_burst_cycle} and total_exp {total_exposure}")
def on_stage(self) -> None:
scan_name = self.parent.scaninfo.scan_msg.content["info"].get("scan_name", "")
if scan_name == "jjf_test":
exp_time = 480e-6#self.parent.scaninfo.exp_time
readout = 20e-6#self.parent.scaninfo.readout_time
total_exposure = exp_time+readout
num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
num_burst_cycle = int(num_burst_cycle * self.parent.scaninfo.exp_time/total_exposure)
self.parent.set_channels("width", exp_time)
self.parent.set_channels("delay", 0.0)
logger.info(f"{self.parent.name}: On stage with n_burst: {num_burst_cycle} and total_exp {total_exposure}")
self.parent.burst_enable(num_burst_cycle, 0, total_exposure, config="first")
def on_trigger(self) -> None:
"""Method to be executed upon trigger"""
@@ -226,10 +245,12 @@ class DDGSetup(DDGCustomMixin):
self.parent.trigger_shot.put(1)
scan_name = self.parent.scaninfo.scan_msg.content["info"].get("scan_name", "")
if scan_name == "jjf_test":
exp_time = self.parent.scaninfo.exp_time
readout = self.parent.scaninfo.readout_time
num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
exp_time = 480e-6#self.parent.scaninfo.exp_time
readout = 20e-6#self.parent.scaninfo.readout_time
total_exposure = exp_time+readout
num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
num_burst_cycle = int(num_burst_cycle * self.parent.scaninfo.exp_time/total_exposure)
cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["cycles"]
#time.sleep(num_burst_cycle*total_exposure)
def check_ddg()->int:
self.parent.trigger_burst_readout.put(1)

0
csaxs_bec/devices/jungfraujoch/__init__.py Normal file
View File

166
csaxs_bec/devices/jungfraujoch/eiger_jungfrau_joch.py Normal file
View File

@@ -0,0 +1,166 @@
import enum
import os
import threading
import time
from typing import Any
import numpy as np
import openapi_client
from bec_lib.logger import bec_logger
from openapi_client.models.dataset_settings import DatasetSettings
from ophyd import ADComponent as ADCpt
from ophyd import Device, EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
from std_daq_client import StdDaqClient
from ophyd_devices.interfaces.base_classes.psi_detector_base import (
CustomDetectorMixin,
PSIDetectorBase,
)
logger = bec_logger.logger
class EigerError(Exception):
"""Base class for exceptions in this module."""
class EigerTimeoutError(EigerError):
"""Raised when the Eiger does not respond in time."""
class DetectorState(str, enum.Enum):
"""Detector states for Eiger9M detector"""
IDLE = "idle"
class ErrorState(int, enum.Enum):
""" Error State in the detector"""
ERROR_STATUS_WAITING = 504
class Eiger9MJungfrauJochSetup(CustomDetectorMixin):
"""Eiger setup class
Parent class: CustomDetectorMixin
"""
def __init__(self, *args, parent: Device = None, **kwargs) -> None:
super().__init__(*args, parent=parent, **kwargs)
# kwargs["host"] =
# kwargs["port"] =
configuration = openapi_client.Configuration(host="http://sls-jfjoch-001:8080")
api_client = openapi_client.ApiClient(configuration)
self.api = openapi_client.DefaultApi(api_client)
self.actually_init()
def actually_init(self):
status = self.get_daq_status()
if status != DetectorState.IDLE:
self.api.initialize_post()
self.actually_wait_till_done()
def actually_wait_till_done(self):
while True:
try:
done = self.api.wait_till_done_post()
except Exception as e: #TODO: be more specific
if e.status != ErrorState.ERROR_STATUS_WAITING:
print(e)
raise e
else:
#TODO: use number_of_triggers_left for progress...
if done is None: # seems we get None instead of: status == 200
return
time.sleep(0.1) #TODO
def get_daq_status(self):
return self.api.status_get().state
def on_stage(self):
scan_name = self.parent.scaninfo.scan_msg.content["info"].get("scan_name", "")
if scan_name != "jjf_test":
return
exp_time = self.parent.scaninfo.exp_time
readout = self.parent.scaninfo.readout_time
num_burst_cycle = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["num_points"]
cycles = self.parent.scaninfo.scan_msg.content["info"]["kwargs"]["cycles"]
total_points = num_burst_cycle * cycles
dataset_settings = DatasetSettings(
image_time_us = int(exp_time*1e6),
ntrigger=total_points,
beam_x_pxl = 0,
beam_y_pxl = 0,
detector_distance_mm = 100,
incident_energy_ke_v = 10.00,
)
self.api.start_post(dataset_settings=dataset_settings)
def on_unstage(self):
pass
def on_complete(self):
logger.info("Starting complete for {self.parent.name}")
scan_name = self.parent.scaninfo.scan_msg.content["info"].get("scan_name", "")
if scan_name != "jjf_test":
return
def wait_till_done_post():
try:
done = self.api.wait_till_done_post(timeout=1)
except Exception as e: #TODO: be more specific
return False
else:
#TODO: use number_of_triggers_left for progress...
if done is None: # seems we get None instead of: status == 200
return True
return True
status = self.wait_with_status(signal_conditions=[(wait_till_done_post, True)], check_stopped=True,timeout=10)
return status
def on_stop(self):
self.api.cancel_post()
class TriggerSource(int, enum.Enum):
"""Trigger signals for Eiger9M detector"""
AUTO = 0
TRIGGER = 1
GATING = 2
BURST_TRIGGER = 3
class Eiger9McSAXS(PSIDetectorBase):
"""
Eiger9M detector for CSAXS
Parent class: PSIDetectorBase
class attributes:
custom_prepare_cls (FalconSetup) : Custom detector setup class for cSAXS,
inherits from CustomDetectorMixin
PSIDetectorBase.set_min_readout (float) : Minimum readout time for the detector
Various EpicsPVs for controlling the detector
"""
# Specify which functions are revealed to the user in BEC client
USER_ACCESS = []
# specify Setup class
custom_prepare_cls = Eiger9MJungfrauJochSetup
# specify minimum readout time for detector and timeout for checks after unstage
MIN_READOUT = 20e-6
TIMEOUT_FOR_SIGNALS = 5
# specify class attributes
if __name__ == "__main__":
eiger = Eiger9McSAXS(name="eiger", prefix="X12SA-ES-EIGER9M:")
eiger.custom_prepare.client.init()

84
csaxs_bec/devices/jungfraujoch/jungfraujoch_client.py Normal file
View File

@@ -0,0 +1,84 @@
import string
from time import sleep
import openapi_client
ERROR_STATUS_WAITING = 504
STATUS_IDLE = "Idle"
# allow / to enable creation of subfolders
ALLOWED_CHARS = set(
string.ascii_letters + string.digits + "_-+/"
)
def character_cleanup(s, default="_", allowed=ALLOWED_CHARS):
return "".join(i if i in allowed else default for i in s)
class JFJClient:
def __init__(self, host):
configuration = openapi_client.Configuration(host=host)
api_client = openapi_client.ApiClient(configuration)
self.api = openapi_client.DefaultApi(api_client)
self.actually_init()
def actually_init(self):
status = self.get_daq_status()
if status != STATUS_IDLE:
self.api.initialize_post()
self.actually_wait_till_done()
status = self.get_daq_status()
if status != STATUS_IDLE:
raise RuntimeError(f"status is not {STATUS_IDLE} but: {status}")
def actually_wait_till_done(self):
while True:
try:
done = self.api.wait_till_done_post()
except Exception as e: #TODO: be more specific
if e.status != ERROR_STATUS_WAITING:
print(e)
raise e
else:
#TODO: use number_of_triggers_left for progress...
if done is None: # seems we get None instead of: status == 200
return
sleep(0.1) #TODO
def get_daq_status(self):
return self.api.status_get().state
def get_detector_status(self):
return self.api.detector_status_get()#.state #TODO
def get_detectors(self):
return self.api.config_select_detector_get()
def get_detector_config(self):
return self.api.config_detector_get()
def acquire(self, file_prefix, **kwargs):
status = self.get_daq_status()
if status != STATUS_IDLE:
raise RuntimeError(f"status is not {STATUS_IDLE} but: {status}")
file_prefix = character_cleanup(file_prefix)
dataset_settings = openapi_client.DatasetSettings(file_prefix=file_prefix, **kwargs)
self.api.start_post(dataset_settings=dataset_settings)
self.actually_wait_till_done()
def take_pedestal(self):
self.api.pedestal_post()
self.actually_wait_till_done()