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

Merge branch 'csaxs_detector_integration' into 'master'

Browse Source 
Csaxs detector integration

See merge request bec/ophyd_devices!34
This commit is contained in:
wakonig_k 2023-09-07 11:45:32 +02:00
commit 49029649aa
13 changed files with 2028 additions and 157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

479
ophyd_devices/epics/devices/DelayGeneratorDG645.py
View File

@ -1,18 +1,28 @@
# -*- coding: utf-8 -*-
"""
Created on Tue Nov 9 16:12:47 2021
@author: mohacsi_i
"""
import enum
import threading
import time
from typing import Any, List
from ophyd import Device, Component, EpicsSignal, EpicsSignalRO, Kind
from ophyd import PVPositioner, Signal
from ophyd import PVPositioner, Signal, DeviceStatus
from ophyd.pseudopos import (
pseudo_position_argument,
real_position_argument,
PseudoSingle,
PseudoPositioner,
)
from ophyd_devices.utils.socket import data_shape, data_type
from ophyd_devices.utils import bec_utils as bec_utils
from bec_lib.core import bec_logger
from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
logger = bec_logger.logger
class DDGError(Exception):
pass
class DelayStatic(Device):
@ -33,10 +43,18 @@ class DelayStatic(Device):
kind=Kind.config,
)
amplitude = Component(
EpicsSignal, "OutputAmpAI", write_pv="OutputAmpAO", name="amplitude", kind=Kind.config
EpicsSignal,
"OutputAmpAI",
write_pv="OutputAmpAO",
name="amplitude",
kind=Kind.config,
)
offset = Component(
EpicsSignal, "OutputOffsetAI", write_pv="OutputOffsetAO", name="offset", kind=Kind.config
EpicsSignal,
"OutputOffsetAI",
write_pv="OutputOffsetAO",
name="offset",
kind=Kind.config,
)
@ -57,9 +75,6 @@ class DelayPair(PseudoPositioner):
# The pseudo positioner axes
delay = Component(PseudoSingle, limits=(0, 2000.0), name="delay")
width = Component(PseudoSingle, limits=(0, 2000.0), name="pulsewidth")
# The real delay axes
# ch1 = Component(EpicsSignal, "DelayAI", write_pv="DelayAO", name="ch1", put_complete=True, kind=Kind.config)
# ch2 = Component(EpicsSignal, "DelayAI", write_pv="DelayAO", name="ch2", put_complete=True, kind=Kind.config)
ch1 = Component(DummyPositioner, name="ch1")
ch2 = Component(DummyPositioner, name="ch2")
io = Component(DelayStatic, name="io")
@ -85,6 +100,16 @@ class DelayPair(PseudoPositioner):
return self.PseudoPosition(delay=real_pos.ch1, width=real_pos.ch2 - real_pos.ch1)
class TriggerSource(int, enum.Enum):
INTERNAL = 0
EXT_RISING_EDGE = 1
EXT_FALLING_EDGE = 2
SS_EXT_RISING_EDGE = 3
SS_EXT_FALLING_EDGE = 4
SINGLE_SHOT = 5
LINE = 6
class DelayGeneratorDG645(Device):
"""DG645 delay generator
@ -109,8 +134,25 @@ class DelayGeneratorDG645(Device):
current device
"""
state = Component(EpicsSignalRO, "EventStatusLI", name="status_register")
SUB_PROGRESS = "progress"
SUB_VALUE = "value"
_default_sub = SUB_VALUE
USER_ACCESS = [
"set_channels",
"_set_trigger",
"burst_enable",
"burst_disable",
"reload_config",
]
trigger_burst_readout = Component(
EpicsSignal, "EventStatusLI.PROC", name="trigger_burst_readout"
)
burst_cycle_finished = Component(EpicsSignalRO, "EventStatusMBBID.B3", name="read_burst_state")
delay_finished = Component(EpicsSignalRO, "EventStatusMBBID.B2", name="delay_finished")
status = Component(EpicsSignalRO, "StatusSI", name="status")
clear_error = Component(EpicsSignal, "StatusClearBO", name="clear_error")
# Front Panel
channelT0 = Component(DelayStatic, "T0", name="T0")
@ -155,62 +197,423 @@ class DelayGeneratorDG645(Device):
name="trigger_rate",
kind=Kind.config,
)
# Command PVs
# arm = Component(EpicsSignal, "TriggerDelayBO", name="arm", kind=Kind.omitted)
trigger_shot = Component(EpicsSignal, "TriggerDelayBO", name="trigger_shot", kind="config")
# Burst mode
burstMode = Component(
EpicsSignal, "BurstModeBI", write_pv="BurstModeBO", name="burstmode", kind=Kind.config
EpicsSignal,
"BurstModeBI",
write_pv="BurstModeBO",
name="burstmode",
kind=Kind.config,
)
burstConfig = Component(
EpicsSignal, "BurstConfigBI", write_pv="BurstConfigBO", name="burstconfig", kind=Kind.config
EpicsSignal,
"BurstConfigBI",
write_pv="BurstConfigBO",
name="burstconfig",
kind=Kind.config,
)
burstCount = Component(
EpicsSignal, "BurstCountLI", write_pv="BurstCountLO", name="burstcount", kind=Kind.config
EpicsSignal,
"BurstCountLI",
write_pv="BurstCountLO",
name="burstcount",
kind=Kind.config,
)
burstDelay = Component(
EpicsSignal, "BurstDelayAI", write_pv="BurstDelayAO", name="burstdelay", kind=Kind.config
EpicsSignal,
"BurstDelayAI",
write_pv="BurstDelayAO",
name="burstdelay",
kind=Kind.config,
)
burstPeriod = Component(
EpicsSignal, "BurstPeriodAI", write_pv="BurstPeriodAO", name="burstperiod", kind=Kind.config
EpicsSignal,
"BurstPeriodAI",
write_pv="BurstPeriodAO",
name="burstperiod",
kind=Kind.config,
)
delay_burst = Component(
bec_utils.ConfigSignal,
name="delay_burst",
kind="config",
config_storage_name="ddg_config",
)
delta_width = Component(
bec_utils.ConfigSignal,
name="delta_width",
kind="config",
config_storage_name="ddg_config",
)
additional_triggers = Component(
bec_utils.ConfigSignal,
name="additional_triggers",
kind="config",
config_storage_name="ddg_config",
)
polarity = Component(
bec_utils.ConfigSignal,
name="polarity",
kind="config",
config_storage_name="ddg_config",
)
amplitude = Component(
bec_utils.ConfigSignal,
name="amplitude",
kind="config",
config_storage_name="ddg_config",
)
offset = Component(
bec_utils.ConfigSignal,
name="offset",
kind="config",
config_storage_name="ddg_config",
)
thres_trig_level = Component(
bec_utils.ConfigSignal,
name="thres_trig_level",
kind="config",
config_storage_name="ddg_config",
)
set_high_on_exposure = Component(
bec_utils.ConfigSignal,
name="set_high_on_exposure",
kind="config",
config_storage_name="ddg_config",
)
set_high_on_stage = Component(
bec_utils.ConfigSignal,
name="set_high_on_stage",
kind="config",
config_storage_name="ddg_config",
)
set_trigger_source = Component(
bec_utils.ConfigSignal,
name="set_trigger_source",
kind="config",
config_storage_name="ddg_config",
)
trigger_width = Component(
bec_utils.ConfigSignal,
name="trigger_width",
kind="config",
config_storage_name="ddg_config",
)
def __init__(
self,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
device_manager=None,
sim_mode=False,
ddg_config=None,
**kwargs,
):
"""_summary_
Args:
name (_type_): _description_
prefix (str, optional): _description_. Defaults to "".
kind (_type_, optional): _description_. Defaults to None.
read_attrs (_type_, optional): _description_. Defaults to None.
configuration_attrs (_type_, optional): _description_. Defaults to None.
parent (_type_, optional): _description_. Defaults to None.
device_manager (_type_, optional): _description_. Defaults to None.
Signals:
polarity (_type_, optional): _description_. Defaults to None.
amplitude (_type_, optional): _description_. Defaults to None.
offset (_type_, optional): _description_. Defaults to None.
thres_trig_level (_type_, optional): _description_. Defaults to None.
delay_burst (_type_, float): Add delay for triggering in software trigger mode to allow fast shutter to open. Defaults to 0.
delta_width (_type_, float): Add width to fast shutter signal to make sure its open during acquisition. Defaults to 0.
delta_triggers (_type_, int): Add additional triggers to burst mode (mcs card needs +1 triggers per line). Defaults to 0.
set_high_on_exposure
set_high_on_stage
set_trigger_source
"""
self.ddg_config = {
f"{name}_delay_burst": 0,
f"{name}_delta_width": 0,
f"{name}_additional_triggers": 0,
f"{name}_polarity": [1, 1, 1, 1, 1],
f"{name}_amplitude": 4.5,
f"{name}_offset": 0,
f"{name}_thres_trig_level": 2.5,
f"{name}_set_high_on_exposure": False,
f"{name}_set_high_on_stage": False,
f"{name}_set_trigger_source": "SINGLE_SHOT",
f"{name}_trigger_width": None,
}
if ddg_config is not None:
[self.ddg_config.update({f"{name}_{key}": value}) for key, value in ddg_config.items()]
super().__init__(
prefix=prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs,
)
if device_manager is None and not sim_mode:
raise DDGError("Add DeviceManager to initialization or init with sim_mode=True")
self.device_manager = device_manager
if not sim_mode:
self._producer = self.device_manager.producer
else:
self._producer = bec_utils.MockProducer()
self.device_manager = bec_utils.MockDeviceManager()
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self._all_channels = [
"channelT0",
"channelAB",
"channelCD",
"channelEF",
"channelGH",
]
self._all_delay_pairs = ["AB", "CD", "EF", "GH"]
self.wait_for_connection() # Make sure to be connected before talking to PVs
logger.info(f"Current polarity values {self.polarity.get()}")
self.reload_config()
self._ddg_is_okay()
self._stopped = False
def _set_trigger(self, trigger_source: TriggerSource) -> None:
"""Set trigger source to value of list below, or string
Accepts integer 0-6 or TriggerSource.* with *
INTERNAL = 0
EXT_RISING_EDGE = 1
EXT_FALLING_EDGE = 2
SS_EXT_RISING_EDGE = 3
SS_EXT_FALLING_EDGE = 4
SINGLE_SHOT = 5
LINE = 6
"""
value = int(trigger_source)
self.source.put(value)
def _ddg_is_okay(self, raise_on_error=False) -> None:
status = self.status.read()[self.status.name]["value"]
if status != "STATUS OK" and not raise_on_error:
logger.warning(f"DDG returns {status}, trying to clear ERROR")
self.clear_error()
time.sleep(1)
self._ddg_is_okay(rais_on_error=True)
elif status != "STATUS OK":
raise DDGError(f"DDG failed to start with status: {status}")
def set_channels(self, signal: str, value: Any, channels: List = None) -> None:
if not channels:
channels = self._all_channels
for chname in channels:
channel = getattr(self, chname, None)
if not channel:
continue
if signal in channel.component_names:
getattr(channel, signal).set(value)
continue
if "io" in channel.component_names and signal in channel.io.component_names:
getattr(channel.io, signal).set(value)
def _cleanup_ddg(self) -> None:
self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
def reload_config(self) -> None:
for ii, channel in enumerate(self._all_channels):
self.set_channels("polarity", self.polarity.get()[ii], channels=[channel])
# Set polarity for eiger inverted!
# self.set_channels("polarity", 0, channels=["channelAB"])
self.set_channels("amplitude", self.amplitude.get())
self.set_channels("offset", self.offset.get())
# Setup reference
self.set_channels(
"reference",
0,
[f"channel{self._all_delay_pairs[ii]}.ch1" for ii in range(len(self._all_delay_pairs))],
)
for ii in range(len(self._all_delay_pairs)):
self.set_channels(
"reference",
0,
[f"channel{self._all_delay_pairs[ii]}.ch2"],
)
self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
# Set threshold level for ext. pulses
self.level.put(self.thres_trig_level.get())
def _check_burst_cycle(self, status) -> None:
"""Checks burst cycle of delay generator
Force readout, return value from end of burst cycle
"""
while True:
self.trigger_burst_readout.put(1)
if (
self.burst_cycle_finished.read()[self.burst_cycle_finished.name]["value"] == 1
and self.delay_finished.read()[self.delay_finished.name]["value"] == 1
):
self._acquisition_done = True
status.set_finished()
return
if self._stopped == True:
status.set_finished()
break
time.sleep(0.01)
def stop(self, success=False):
"""Stops the DDG"""
self._stopped = True
self._acquisition_done = True
super().stop(success=success)
def stage(self):
"""Trigger the generator by arming to accept triggers"""
# TODO check PV TriggerDelayBO, seems to be a bug in the IOC
# self.arm.write(1).wait()
self.scaninfo.load_scan_metadata()
if self.scaninfo.scan_type == "step":
# define parameters
if self.set_high_on_exposure.get():
self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
num_burst_cycle = 1 + self.additional_triggers.get()
exp_time = self.delta_width.get() + self.scaninfo.frames_per_trigger * (
self.scaninfo.exp_time + self.scaninfo.readout_time
)
total_exposure = exp_time
delay_burst = self.delay_burst.get()
self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
self.set_channels("delay", 0)
# Set burst length to half of the experimental time!
if not self.trigger_width.get():
self.set_channels("width", exp_time)
else:
self.set_channels("width", self.trigger_width.get())
else:
self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
exp_time = self.delta_width.get() + self.scaninfo.exp_time
total_exposure = exp_time + self.scaninfo.readout_time
delay_burst = self.delay_burst.get()
num_burst_cycle = self.scaninfo.frames_per_trigger + self.additional_triggers.get()
# set parameters in DDG
self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
self.set_channels("delay", 0)
# Set burst length to half of the experimental time!
if not self.trigger_width.get():
self.set_channels("width", exp_time)
else:
self.set_channels("width", self.trigger_width.get())
elif self.scaninfo.scan_type == "fly":
if self.set_high_on_exposure.get():
# define parameters
self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
exp_time = (
self.delta_width.get()
+ self.scaninfo.exp_time * self.scaninfo.num_points
+ self.scaninfo.readout_time * (self.scaninfo.num_points - 1)
)
total_exposure = exp_time
delay_burst = self.delay_burst.get()
# self.additional_triggers should be 0 for self.set_high_on_exposure or remove here fully..
num_burst_cycle = 1 + self.additional_triggers.get()
# set parameters in DDG
self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
self.set_channels("delay", 0.0)
# Set burst length to half of the experimental time!
if not self.trigger_width.get():
self.set_channels("width", exp_time)
else:
self.set_channels("width", self.trigger_width.get())
else:
# define parameters
self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
exp_time = self.delta_width.get() + self.scaninfo.exp_time
total_exposure = exp_time + self.scaninfo.readout_time
delay_burst = self.delay_burst.get()
num_burst_cycle = self.scaninfo.num_points + self.additional_triggers.get()
# set parameters in DDG
self.burst_enable(num_burst_cycle, delay_burst, total_exposure, config="first")
self.set_channels("delay", 0.0)
# Set burst length to half of the experimental time!
if not self.trigger_width.get():
self.set_channels("width", exp_time)
else:
self.set_channels("width", self.trigger_width.get())
else:
raise DDGError(f"Unknown scan type {self.scaninfo.scan_type}")
# Check status
self._ddg_is_okay()
logger.info("DDG staged")
super().stage()
def unstage(self):
"""Stop the trigger generator from accepting triggers"""
# self.arm.write(0).wait()
super().stage()
# self._set_trigger(getattr(TriggerSource, self.set_trigger_source.get()))
# Check status
self._ddg_is_okay()
self._stopped = False
self._acquisition_done = False
super().unstage()
def burstEnable(self, count, delay, period, config="all"):
def trigger(self) -> DeviceStatus:
# if self.scaninfo.scan_type == "step":
if self.source.read()[self.source.name]["value"] == int(TriggerSource.SINGLE_SHOT):
self.trigger_shot.put(1)
# status = super().trigger(status=)
status = DeviceStatus(self)
burst_state = threading.Thread(target=self._check_burst_cycle, args=(status,), daemon=True)
burst_state.start()
return status
def burst_enable(self, count, delay, period, config="all"):
"""Enable the burst mode"""
# Validate inputs
count = int(count)
assert count > 0, "Number of bursts must be positive"
assert delay >= 0, "Burst delay must be larger than 0"
assert period > 0, "Burst period must be positive"
assert config in ["all", "first"], "Supported bust configs are 'all' and 'first'"
assert config in [
"all",
"first",
], "Supported bust configs are 'all' and 'first'"
self.burstMode.set(1).wait()
self.burstCount.set(count).wait()
self.burstDelay.set(delay).wait()
self.burstPeriod.set(period).wait()
self.burstMode.put(1)
self.burstCount.put(count)
self.burstDelay.put(delay)
self.burstPeriod.put(period)
if config == "all":
self.burstConfig.set(0).wait()
self.burstConfig.put(0)
elif config == "first":
self.burstConfig.set(1).wait()
self.burstConfig.put(1)
def burstDisable(self):
def burst_disable(self):
"""Disable the burst mode"""
self.burstMode.set(0).wait()
self.burstMode.put(0)
# Automatically connect to test environmenr if directly invoked
if __name__ == "__main__":
dgen = DelayGeneratorDG645("X01DA-PC-DGEN:", name="delayer")
dgen = DelayGeneratorDG645("delaygen:DG1:", name="dgen", sim_mode=True)
# start = time.time()
# dgen.stage()
# dgen.trigger()
# print(f"Time passed for stage and trigger {time.time()-start}s")

9
ophyd_devices/epics/devices/Test.py Normal file
View File

@ -0,0 +1,9 @@
import os
from ophyd_devices.epics.devices.pilatus_csaxs import PilatusCsaxs
os.environ["EPICS_CA_AUTO_ADDR_LIST"] = "NO"
os.environ["EPICS_CA_ADDR_LIST"] = "129.129.122.255 sls-x12sa-cagw.psi.ch:5824"
# pilatus_2 = PilatusCsaxs(name="pilatus_2", prefix="X12SA-ES-PILATUS300K")
# pilatus_2.stage()

8
ophyd_devices/epics/devices/__init__.py
View File

@ -21,3 +21,11 @@ from .specMotors import (
from ophyd import EpicsSignal, EpicsSignalRO, EpicsMotor
from ophyd.sim import SynAxis, SynSignal, SynPeriodicSignal
from ophyd.quadem import QuadEM
# cSAXS
from .epics_motor_ex import EpicsMotorEx
from .mcs_csaxs import McsCsaxs
from .eiger9m_csaxs import Eiger9mCsaxs
from .pilatus_csaxs import PilatusCsaxs
from .falcon_csaxs import FalconCsaxs
from .DelayGeneratorDG645 import DelayGeneratorDG645

69
ophyd_devices/epics/devices/bec_scaninfo_mixin.py Normal file
View File

@ -0,0 +1,69 @@
import os
from bec_lib.core import DeviceManagerBase, BECMessage, MessageEndpoints
from bec_lib.core import bec_logger
logger = bec_logger.logger
class BecScaninfoMixin:
def __init__(self, device_manager: DeviceManagerBase = None, sim_mode=False) -> None:
self.device_manager = device_manager
self.sim_mode = sim_mode
self.scan_msg = None
self.scanID = None
self.bec_info_msg = {
"RID": "mockrid",
"queueID": "mockqueuid",
"scan_number": 1,
"exp_time": 12e-3,
"num_points": 500,
"readout_time": 3e-3,
"scan_type": "fly",
"num_lines": 1,
"frames_per_trigger": 1,
}
def get_bec_info_msg(self) -> None:
return self.bec_info_msg
def change_config(self, bec_info_msg: dict) -> None:
self.bec_info_msg = bec_info_msg
def _get_current_scan_msg(self) -> BECMessage.ScanStatusMessage:
if not self.sim_mode:
# TODO what if no scan info is there yet!
msg = self.device_manager.producer.get(MessageEndpoints.scan_status())
return BECMessage.ScanStatusMessage.loads(msg)
return BECMessage.ScanStatusMessage(
scanID="1",
status={},
info=self.bec_info_msg,
)
def get_username(self) -> str:
if not self.sim_mode:
return self.device_manager.producer.get(MessageEndpoints.account()).decode()
return os.getlogin()
def load_scan_metadata(self) -> None:
self.scan_msg = scan_msg = self._get_current_scan_msg()
logger.info(f"{self.scan_msg}")
try:
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.frames_per_trigger = scan_msg.content["info"]["frames_per_trigger"]
self.num_points = scan_msg.content["info"]["num_points"]
self.scan_type = scan_msg.content["info"].get("scan_type", "step")
self.readout_time = scan_msg.content["info"]["readout_time"]
except Exception as exc:
logger.error(f"Failed to load scan metadata: {exc}.")
self.username = self.get_username()

346
ophyd_devices/epics/devices/eiger9m_csaxs.py Normal file
View File

@ -0,0 +1,346 @@
import enum
import time
from typing import Any, List
import numpy as np
from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
from ophyd import DetectorBase, Device
from ophyd import ADComponent as ADCpt
from bec_lib.core import BECMessage, MessageEndpoints
from bec_lib.core.file_utils import FileWriterMixin
from bec_lib.core import bec_logger
from ophyd_devices.utils import bec_utils as bec_utils
from std_daq_client import StdDaqClient
from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
logger = bec_logger.logger
class EigerError(Exception):
pass
class SlsDetectorCam(Device):
detector_type = ADCpt(EpicsSignalRO, "DetectorType_RBV")
setting = ADCpt(EpicsSignalWithRBV, "Setting")
delay_time = ADCpt(EpicsSignalWithRBV, "DelayTime")
threshold_energy = ADCpt(EpicsSignalWithRBV, "ThresholdEnergy")
beam_energy = ADCpt(EpicsSignalWithRBV, "BeamEnergy")
enable_trimbits = ADCpt(EpicsSignalWithRBV, "Trimbits")
bit_depth = ADCpt(EpicsSignalWithRBV, "BitDepth")
num_gates = ADCpt(EpicsSignalWithRBV, "NumGates")
num_cycles = ADCpt(EpicsSignalWithRBV, "NumCycles")
num_frames = ADCpt(EpicsSignalWithRBV, "NumFrames")
timing_mode = ADCpt(EpicsSignalWithRBV, "TimingMode")
trigger_software = ADCpt(EpicsSignal, "TriggerSoftware")
high_voltage = ADCpt(EpicsSignalWithRBV, "HighVoltage")
# Receiver and data callback
receiver_mode = ADCpt(EpicsSignalWithRBV, "ReceiverMode")
receiver_stream = ADCpt(EpicsSignalWithRBV, "ReceiverStream")
enable_data = ADCpt(EpicsSignalWithRBV, "UseDataCallback")
missed_packets = ADCpt(EpicsSignalRO, "ReceiverMissedPackets_RBV")
# Direct settings access
setup_file = ADCpt(EpicsSignal, "SetupFile")
load_setup = ADCpt(EpicsSignal, "LoadSetup")
command = ADCpt(EpicsSignal, "Command")
# Mythen 3
counter_mask = ADCpt(EpicsSignalWithRBV, "CounterMask")
counter1_threshold = ADCpt(EpicsSignalWithRBV, "Counter1Threshold")
counter2_threshold = ADCpt(EpicsSignalWithRBV, "Counter2Threshold")
counter3_threshold = ADCpt(EpicsSignalWithRBV, "Counter3Threshold")
gate1_delay = ADCpt(EpicsSignalWithRBV, "Gate1Delay")
gate1_width = ADCpt(EpicsSignalWithRBV, "Gate1Width")
gate2_delay = ADCpt(EpicsSignalWithRBV, "Gate2Delay")
gate2_width = ADCpt(EpicsSignalWithRBV, "Gate2Width")
gate3_delay = ADCpt(EpicsSignalWithRBV, "Gate3Delay")
gate3_width = ADCpt(EpicsSignalWithRBV, "Gate3Width")
# Moench
json_frame_mode = ADCpt(EpicsSignalWithRBV, "JsonFrameMode")
json_detector_mode = ADCpt(EpicsSignalWithRBV, "JsonDetectorMode")
# fixes due to missing PVs from CamBase
acquire = ADCpt(EpicsSignal, "Acquire")
detector_state = ADCpt(EpicsSignalRO, "DetectorState_RBV")
class TriggerSource(int, enum.Enum):
AUTO = 0
TRIGGER = 1
GATING = 2
BURST_TRIGGER = 3
class DetectorState(int, enum.Enum):
IDLE = 0
ERROR = 1
WAITING = 2
FINISHED = 3
TRANSMITTING = 4
RUNNING = 5
STOPPED = 6
STILL_WAITING = 7
INITIALIZING = 8
DISCONNECTED = 9
ABORTED = 10
class Eiger9mCsaxs(DetectorBase):
"""Eiger 9M detector for CSAXS
Parent class: DetectorBase
Device class: SlsDetectorCam
Attributes:
name str: 'eiger'
prefix (str): PV prefix (X12SA-ES-EIGER9M:)
"""
cam = ADCpt(SlsDetectorCam, "cam1:")
def __init__(
self,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
device_manager=None,
sim_mode=False,
**kwargs,
):
super().__init__(
prefix=prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs,
)
self._stopped = False
if device_manager is None and not sim_mode:
raise EigerError("Add DeviceManager to initialization or init with sim_mode=True")
self.name = name
self.wait_for_connection() # Make sure to be connected before talking to PVs
if not sim_mode:
from bec_lib.core.bec_service import SERVICE_CONFIG
self.device_manager = device_manager
self._producer = self.device_manager.producer
self.service_cfg = SERVICE_CONFIG.config["service_config"]["file_writer"]
else:
self._producer = bec_utils.MockProducer()
self.device_manager = bec_utils.MockDeviceManager()
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self.scaninfo.load_scan_metadata()
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"}
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self.scaninfo.load_scan_metadata()
# TODO
self.filepath = ""
self.filewriter = FileWriterMixin(self.service_cfg)
self.reduce_readout = 1e-3 # 3 ms
self.triggermode = 0 # 0 : internal, scan must set this if hardware triggered
self._init_eiger9m()
self._init_standard_daq()
# self.mokev = self.device_manager.devices.mokev.read()[
# self.device_manager.devices.mokev.name
# ]["value"]
def _init_eiger9m(self) -> None:
"""Init parameters for Eiger 9m"""
self._set_trigger(TriggerSource.GATING)
self.cam.acquire.set(0)
def _update_std_cfg(self, cfg_key: str, value: Any) -> None:
cfg = self.std_client.get_config()
old_value = cfg.get(cfg_key)
if old_value is None:
raise EigerError(
f"Tried to change entry for key {cfg_key} in std_config that does not exist"
)
if not isinstance(value, type(old_value)):
raise EigerError(
f"Type of new value {type(value)}:{value} does not match old value {type(old_value)}:{old_value}"
)
cfg.update({cfg_key: value})
logger.info(f"Updated std_daq config for key {cfg_key} from {old_value} to {value}")
def _init_standard_daq(self) -> None:
self.std_rest_server_url = "http://xbl-daq-29:5000"
self.std_client = StdDaqClient(url_base=self.std_rest_server_url)
self.std_client.stop_writer()
timeout = 0
self._update_std_cfg("writer_user_id", int(self.scaninfo.username.strip(" e")))
time.sleep(1)
while not self.std_client.get_status()["state"] == "READY":
time.sleep(0.1)
timeout = timeout + 0.1
logger.info("Waiting for std_daq init.")
if timeout > 2:
if not self.std_client.get_status()["state"]:
raise EigerError(
f"Std client not in READY state, returns: {self.std_client.get_status()}"
)
else:
return
def _prep_det(self) -> None:
self._set_det_threshold()
self._set_acquisition_params()
self._set_trigger(TriggerSource.GATING)
def _set_det_threshold(self) -> None:
# threshold_energy PV exists on Eiger 9M?
factor = 1
if self.cam.threshold_energy._metadata["units"] == "eV":
factor = 1000
setp_energy = int(self.mokev * factor)
energy = self.cam.beam_energy.read()[self.cam.beam_energy.name]["value"]
if setp_energy != energy:
self.cam.beam_energy.set(setp_energy) # .wait()
threshold = self.cam.threshold_energy.read()[self.cam.threshold_energy.name]["value"]
if not np.isclose(setp_energy / 2, threshold, rtol=0.05):
self.cam.threshold_energy.set(setp_energy / 2) # .wait()
def _set_acquisition_params(self) -> None:
# self.cam.acquire_time.set(self.scaninfo.exp_time)
# Set acquisition parameters slightly shorter then cycle
# self.cam.acquire_period.set(
# self.scaninfo.exp_time + (self.scaninfo.readout_time - self.reduce_readout)
# )
self.cam.num_cycles.set(int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger))
self.cam.num_frames.set(1)
def _set_trigger(self, trigger_source: TriggerSource) -> None:
"""Set trigger source for the detector, either directly to value or TriggerSource.* with
AUTO = 0
TRIGGER = 1
GATING = 2
BURST_TRIGGER = 3
"""
value = int(trigger_source)
self.cam.timing_mode.set(value)
def _prep_file_writer(self) -> None:
self.filepath = self.filewriter.compile_full_filename(
self.scaninfo.scan_number, f"{self.name}.h5", 1000, 5, True
)
# self._close_file_writer()
logger.info(f" std_daq output filepath {self.filepath}")
try:
self.std_client.start_writer_async(
{
"output_file": self.filepath,
"n_images": int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger),
}
)
except Exception as exc:
time.sleep(5)
if self.std_client.get_status()["state"] == "READY":
raise EigerError(f"Timeout of start_writer_async with {exc}")
while True:
det_ctrl = self.std_client.get_status()["acquisition"]["state"]
if det_ctrl == "WAITING_IMAGES":
break
time.sleep(0.005)
def _close_file_writer(self) -> None:
self.std_client.stop_writer()
pass
def stage(self) -> List[object]:
"""stage the detector and file writer"""
self.scaninfo.load_scan_metadata()
self.mokev = self.device_manager.devices.mokev.obj.read()[
self.device_manager.devices.mokev.name
]["value"]
self._prep_det()
logger.info("Waiting for std daq to be armed")
self._prep_file_writer()
logger.info("std_daq is ready")
msg = BECMessage.FileMessage(file_path=self.filepath, done=False)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
msg.dumps(),
)
msg = BECMessage.FileMessage(file_path=self.filepath, done=False)
self._producer.set_and_publish(
MessageEndpoints.file_event(self.name),
msg.dumps(),
)
self.arm_acquisition()
logger.info("Waiting for Eiger9m to be armed")
while True:
det_ctrl = self.cam.detector_state.read()[self.cam.detector_state.name]["value"]
if det_ctrl == int(DetectorState.RUNNING):
break
if self._stopped == True:
break
time.sleep(0.005)
logger.info("Eiger9m is armed")
self._stopped = False
return super().stage()
def unstage(self) -> List[object]:
"""unstage the detector and file writer"""
logger.info("Waiting for Eiger9M to return from acquisition")
while True:
det_ctrl = self.cam.acquire.read()[self.cam.acquire.name]["value"]
if det_ctrl == 0:
break
if self._stopped == True:
break
time.sleep(0.005)
logger.info("Eiger9M finished")
logger.info("Waiting for std daq to receive images")
while True:
det_ctrl = self.std_client.get_status()["acquisition"]["state"]
# TODO if no writing was performed before
if det_ctrl == "FINISHED":
break
if self._stopped == True:
break
time.sleep(0.005)
logger.info("Std_daq finished")
# Message to BEC
state = True
msg = BECMessage.FileMessage(file_path=self.filepath, done=True, successful=state)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
msg.dumps(),
)
self._stopped = False
return super().unstage()
def arm_acquisition(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._close_file_writer()
super().stop(success=success)
self._stopped = True
if __name__ == "__main__":
eiger = Eiger9mCsaxs(name="eiger", prefix="X12SA-ES-EIGER9M:", sim_mode=True)

45
ophyd_devices/epics/devices/epics_motor_ex.py Normal file
View File

@ -0,0 +1,45 @@
from ophyd import Component as Cpt, EpicsSignal, EpicsMotor
class EpicsMotorEx(EpicsMotor):
"""Extend EpicsMotor with extra configuration fields."""
# configuration
motor_resolution = Cpt(EpicsSignal, ".MRES", kind="config", auto_monitor=True)
base_velocity = Cpt(EpicsSignal, ".VBAS", kind="config", auto_monitor=True)
backlash_distance = Cpt(EpicsSignal, ".BDST", kind="config", auto_monitor=True)
def __init__(
self,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
**kwargs
):
# get configuration attributes from kwargs and then remove them
attrs = {}
for key, value in kwargs.items():
if hasattr(EpicsMotorEx, key) and isinstance(getattr(EpicsMotorEx, key), Cpt):
attrs[key] = value
for key in attrs:
kwargs.pop(key)
super().__init__(
prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs
)
# set configuration attributes
for key, value in attrs.items():
# print out attributes that are being configured
print("setting ", key, "=", value)
getattr(self, key).put(value)

262
ophyd_devices/epics/devices/falcon_csaxs.py Normal file
View File

@ -0,0 +1,262 @@
import enum
import os
import time
from typing import List
from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV, Component as Cpt, Device
from ophyd.mca import EpicsMCARecord
from ophyd.areadetector.plugins import HDF5Plugin_V21, FilePlugin_V22
from bec_lib.core.file_utils import FileWriterMixin
from bec_lib.core import MessageEndpoints, BECMessage
from bec_lib.core import bec_logger
from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
from ophyd_devices.utils import bec_utils
logger = bec_logger.logger
class FalconError(Exception):
pass
class DetectorState(int, enum.Enum):
DONE = 0
ACQUIRING = 1
class EpicsDXPFalcon(Device):
"""All high-level DXP parameters for each channel"""
elapsed_live_time = Cpt(EpicsSignal, "ElapsedLiveTime")
elapsed_real_time = Cpt(EpicsSignal, "ElapsedRealTime")
elapsed_trigger_live_time = Cpt(EpicsSignal, "ElapsedTriggerLiveTime")
# Energy Filter PVs
energy_threshold = Cpt(EpicsSignalWithRBV, "DetectionThreshold")
min_pulse_separation = Cpt(EpicsSignalWithRBV, "MinPulsePairSeparation")
detection_filter = Cpt(EpicsSignalWithRBV, "DetectionFilter", string=True)
scale_factor = Cpt(EpicsSignalWithRBV, "ScaleFactor")
risetime_optimisation = Cpt(EpicsSignalWithRBV, "RisetimeOptimization")
# Misc PVs
detector_polarity = Cpt(EpicsSignalWithRBV, "DetectorPolarity")
decay_time = Cpt(EpicsSignalWithRBV, "DecayTime")
current_pixel = Cpt(EpicsSignalRO, "CurrentPixel")
class FalconHDF5Plugins(Device): # HDF5Plugin_V21, FilePlugin_V22):
capture = Cpt(EpicsSignalWithRBV, "Capture")
enable = Cpt(EpicsSignalWithRBV, "EnableCallbacks", string=True, kind="config")
xml_file_name = Cpt(EpicsSignalWithRBV, "XMLFileName", string=True, kind="config")
lazy_open = Cpt(EpicsSignalWithRBV, "LazyOpen", string=True, doc="0='No' 1='Yes'")
temp_suffix = Cpt(EpicsSignalWithRBV, "TempSuffix", string=True)
# file_path = Cpt(
# EpicsSignalWithRBV, "FilePath", string=True, kind="config", path_semantics="posix"
# )
file_path = Cpt(EpicsSignalWithRBV, "FilePath", string=True, kind="config")
file_name = Cpt(EpicsSignalWithRBV, "FileName", string=True, kind="config")
file_template = Cpt(EpicsSignalWithRBV, "FileTemplate", string=True, kind="config")
num_capture = Cpt(EpicsSignalWithRBV, "NumCapture", kind="config")
file_write_mode = Cpt(EpicsSignalWithRBV, "FileWriteMode", kind="config")
capture = Cpt(EpicsSignalWithRBV, "Capture")
class FalconCsaxs(Device):
"""FalxonX1 with HDF5 writer"""
dxp = Cpt(EpicsDXPFalcon, "dxp1:")
mca = Cpt(EpicsMCARecord, "mca1")
hdf5 = Cpt(FalconHDF5Plugins, "HDF1:")
# Control
stop_all = Cpt(EpicsSignal, "StopAll")
erase_all = Cpt(EpicsSignal, "EraseAll")
start_all = Cpt(EpicsSignal, "StartAll")
state = Cpt(EpicsSignal, "Acquiring")
# Preset options
preset_mode = Cpt(EpicsSignal, "PresetMode") # 0 No preset 1 Real time 2 Events 3 Triggers
preset_real = Cpt(EpicsSignal, "PresetReal")
preset_events = Cpt(EpicsSignal, "PresetEvents")
preset_triggers = Cpt(EpicsSignal, "PresetTriggers")
# read-only diagnostics
triggers = Cpt(EpicsSignalRO, "MaxTriggers", lazy=True)
events = Cpt(EpicsSignalRO, "MaxEvents", lazy=True)
input_count_rate = Cpt(EpicsSignalRO, "MaxInputCountRate", lazy=True)
output_count_rate = Cpt(EpicsSignalRO, "MaxOutputCountRate", lazy=True)
# Mapping control
collect_mode = Cpt(EpicsSignal, "CollectMode") # 0 MCA spectra, 1 MCA mapping
pixel_advance_mode = Cpt(EpicsSignal, "PixelAdvanceMode")
ignore_gate = Cpt(EpicsSignal, "IgnoreGate")
input_logic_polarity = Cpt(EpicsSignal, "InputLogicPolarity")
auto_pixels_per_buffer = Cpt(EpicsSignal, "AutoPixelsPerBuffer")
pixels_per_buffer = Cpt(EpicsSignal, "PixelsPerBuffer")
pixels_per_run = Cpt(EpicsSignal, "PixelsPerRun")
# HDF5
def __init__(
self,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
device_manager=None,
sim_mode=False,
**kwargs,
):
super().__init__(
prefix=prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs,
)
if device_manager is None and not sim_mode:
raise FalconError("Add DeviceManager to initialization or init with sim_mode=True")
self._stopped = False
self.name = name
self.wait_for_connection() # Make sure to be connected before talking to PVs
if not sim_mode:
from bec_lib.core.bec_service import SERVICE_CONFIG
self.device_manager = device_manager
self._producer = self.device_manager.producer
self.service_cfg = SERVICE_CONFIG.config["service_config"]["file_writer"]
else:
self._producer = bec_utils.MockProducer()
self.device_manager = bec_utils.MockDeviceManager()
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self.scaninfo.load_scan_metadata()
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"}
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self.scaninfo.load_scan_metadata()
self.filewriter = FileWriterMixin(self.service_cfg)
self.readout = 0.003 # 3 ms
self._value_pixel_per_buffer = 1 # 16
self._clean_up()
self._init_hdf5_saving()
self._init_mapping_mode()
def _clean_up(self) -> None:
"""Clean up"""
self.hdf5.capture.put(0)
self.stop_all.put(1)
self.erase_all.put(1)
def _init_hdf5_saving(self) -> None:
"""Set up hdf5 save parameters"""
self.hdf5.enable.put(1) # EnableCallbacks
self.hdf5.xml_file_name.put("layout.xml") # Points to hardcopy of HDF5 Layout xml file
self.hdf5.lazy_open.put(1) # Yes -> To be checked how to add FilePlugin_V21+
self.hdf5.temp_suffix.put("temps") # -> To be checked how to add FilePlugin_V22+
def _init_mapping_mode(self) -> None:
"""Set up mapping mode params"""
self.collect_mode.put(1) # 1 MCA Mapping, 0 MCA Spectrum
self.preset_mode.put(1) # 1 Realtime
self.input_logic_polarity.put(0) # 0 Normal, 1 Inverted
self.pixel_advance_mode.put(1) # 0 User, 1 Gate, 2 Sync
self.ignore_gate.put(1) # 1 Yes
self.auto_pixels_per_buffer.put(0) # 0 Manual 1 Auto
self.pixels_per_buffer.put(16) #
def _prep_det(self) -> None:
"""Prepare detector for acquisition"""
self.collect_mode.put(1)
self.preset_real.put(self.scaninfo.exp_time)
self.pixels_per_run.put(int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger))
self.auto_pixels_per_buffer.put(0)
self.pixels_per_buffer.put(self._value_pixel_per_buffer)
def _prep_file_writer(self) -> None:
"""Prep HDF5 weriting"""
# TODO creta filename and destination path from filepath
self.destination_path = self.filewriter.compile_full_filename(
self.scaninfo.scan_number, f"{self.name}.h5", 1000, 5, True
)
# self.hdf5.file_path.set(self.destination_path)
file_path, file_name = os.path.split(self.destination_path)
self.hdf5.file_path.put(file_path)
self.hdf5.file_name.put(file_name)
self.hdf5.file_template.put(f"%s%s")
self.hdf5.num_capture.put(self.scaninfo.num_points // self._value_pixel_per_buffer + 1)
self.hdf5.file_write_mode.put(2)
self.hdf5.capture.put(1)
def stage(self) -> List[object]:
"""stage the detector and file writer"""
# TODO clean up needed?
# self._clean_up()
self.scaninfo.load_scan_metadata()
self.mokev = self.device_manager.devices.mokev.obj.read()[
self.device_manager.devices.mokev.name
]["value"]
logger.info("Waiting for pilatus2 to be armed")
self._prep_det()
logger.info("Pilatus2 armed")
logger.info("Waiting for pilatus2 zmq stream to be ready")
self._prep_file_writer()
logger.info("Pilatus2 zmq ready")
msg = BECMessage.FileMessage(file_path=self.destination_path, done=False)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
msg.dumps(),
)
self.arm_acquisition()
logger.info("Waiting for Falcon to be armed")
while True:
det_ctrl = self.state.read()[self.state.name]["value"]
if det_ctrl == int(DetectorState.ACQUIRING):
break
if self._stopped == True:
break
time.sleep(0.005)
logger.info("Falcon is armed")
self._stopped = False
return super().stage()
def arm_acquisition(self) -> None:
self.start_all.put(1)
def unstage(self) -> List[object]:
logger.info("Waiting for Falcon to return from acquisition")
while True:
det_ctrl = self.state.read()[self.state.name]["value"]
if det_ctrl == int(DetectorState.DONE):
break
if self._stopped == True:
break
time.sleep(0.005)
logger.info("Falcon done")
# TODO needed?
self._clean_up()
state = True
msg = BECMessage.FileMessage(file_path=self.destination_path, done=True, successful=state)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scaninfo.metadata["scanID"], self.name),
msg.dumps(),
)
self._stopped = False
return super().unstage()
def stop(self, *, success=False) -> None:
"""Stop the scan, with camera and file writer"""
self._clean_up()
super().stop(success=success)
self._stopped = True
if __name__ == "__main__":
falcon = FalconCsaxs(name="falcon", prefix="X12SA-SITORO:", sim_mode=True)

383
ophyd_devices/epics/devices/mcs_csaxs.py Normal file
View File

@ -0,0 +1,383 @@
import enum
import threading
import time
from typing import Any, List
import numpy as np
from ophyd import EpicsSignal, EpicsSignalRO
from ophyd import EpicsSignal, EpicsSignalRO, Component as Cpt, Device
from ophyd.mca import EpicsMCARecord
from ophyd.scaler import ScalerCH
from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
from ophyd_devices.utils import bec_utils
from bec_lib.core import BECMessage, MessageEndpoints
from bec_lib.core.file_utils import FileWriterMixin
from collections import defaultdict
from bec_lib.core import bec_logger, threadlocked
logger = bec_logger.logger
class MCSError(Exception):
pass
class TriggerSource(int, enum.Enum):
MODE0 = 0
MODE1 = 1
MODE2 = 2
MODE3 = 3
MODE4 = 4
MODE5 = 5
MODE6 = 6
class ChannelAdvance(int, enum.Enum):
INTERNAL = 0
EXTERNAL = 1
class ReadoutMode(int, enum.Enum):
PASSIVE = 0
EVENT = 1
IO_INTR = 2
FREQ_0_1HZ = 3
FREQ_0_2HZ = 4
FREQ_0_5HZ = 5
FREQ_1HZ = 6
FREQ_2HZ = 7
FREQ_5HZ = 8
FREQ_10HZ = 9
FREQ_100HZ = 10
class SIS38XX(Device):
"""SIS38XX control"""
# Acquisition
erase_all = Cpt(EpicsSignal, "EraseAll")
erase_start = Cpt(EpicsSignal, "EraseStart", trigger_value=1)
start_all = Cpt(EpicsSignal, "StartAll")
stop_all = Cpt(EpicsSignal, "StopAll")
acquiring = Cpt(EpicsSignal, "Acquiring")
preset_real = Cpt(EpicsSignal, "PresetReal")
elapsed_real = Cpt(EpicsSignal, "ElapsedReal")
read_mode = Cpt(EpicsSignal, "ReadAll.SCAN")
read_all = Cpt(EpicsSignal, "DoReadAll.VAL", trigger_value=1)
num_use_all = Cpt(EpicsSignal, "NuseAll")
current_channel = Cpt(EpicsSignal, "CurrentChannel")
dwell = Cpt(EpicsSignal, "Dwell")
channel_advance = Cpt(EpicsSignal, "ChannelAdvance")
count_on_start = Cpt(EpicsSignal, "CountOnStart")
software_channel_advance = Cpt(EpicsSignal, "SoftwareChannelAdvance")
channel1_source = Cpt(EpicsSignal, "Channel1Source")
prescale = Cpt(EpicsSignal, "Prescale")
enable_client_wait = Cpt(EpicsSignal, "EnableClientWait")
client_wait = Cpt(EpicsSignal, "ClientWait")
acquire_mode = Cpt(EpicsSignal, "AcquireMode")
mux_output = Cpt(EpicsSignal, "MUXOutput")
user_led = Cpt(EpicsSignal, "UserLED")
input_mode = Cpt(EpicsSignal, "InputMode")
input_polarity = Cpt(EpicsSignal, "InputPolarity")
output_mode = Cpt(EpicsSignal, "OutputMode")
output_polarity = Cpt(EpicsSignal, "OutputPolarity")
model = Cpt(EpicsSignalRO, "Model", string=True)
firmware = Cpt(EpicsSignalRO, "Firmware")
max_channels = Cpt(EpicsSignalRO, "MaxChannels")
class McsCsaxs(SIS38XX):
SUB_PROGRESS = "progress"
SUB_VALUE = "value"
_default_sub = SUB_VALUE
# scaler = Cpt(ScalerCH, "scaler1")
# mca2 = Cpt(EpicsMCARecord, "mca2")
mca1 = Cpt(EpicsSignalRO, "mca1.VAL", auto_monitor=True)
mca3 = Cpt(EpicsSignalRO, "mca3.VAL", auto_monitor=True)
mca4 = Cpt(EpicsSignalRO, "mca4.VAL", auto_monitor=True)
# mca5 = Cpt(EpicsMCARecord, "mca5")
# mca6 = Cpt(EpicsMCARecord, "mca6")
# mca7 = Cpt(EpicsMCARecord, "mca7")
# mca8 = Cpt(EpicsMCARecord, "mca8")
# mca9 = Cpt(EpicsMCARecord, "mca9")
# mca10 = Cpt(EpicsMCARecord, "mca10")
# mca11 = Cpt(EpicsMCARecord, "mca11")
# mca12 = Cpt(EpicsMCARecord, "mca12")
# mca13 = Cpt(EpicsMCARecord, "mca13")
# mca14 = Cpt(EpicsMCARecord, "mca14")
# mca15 = Cpt(EpicsMCARecord, "mca15")
# mca16 = Cpt(EpicsMCARecord, "mca16")
# mca17 = Cpt(EpicsMCARecord, "mca17")
# mca18 = Cpt(EpicsMCARecord, "mca18")
# mca19 = Cpt(EpicsMCARecord, "mca19")
# mca20 = Cpt(EpicsMCARecord, "mca20")
# mca21 = Cpt(EpicsMCARecord, "mca21")
# mca22 = Cpt(EpicsMCARecord, "mca22")
# mca23 = Cpt(EpicsMCARecord, "mca23")
# mca24 = Cpt(EpicsMCARecord, "mca24")
# mca25 = Cpt(EpicsMCARecord, "mca25")
# mca26 = Cpt(EpicsMCARecord, "mca26")
# mca27 = Cpt(EpicsMCARecord, "mca27")
# mca28 = Cpt(EpicsMCARecord, "mca28")
# mca29 = Cpt(EpicsMCARecord, "mca29")
# mca30 = Cpt(EpicsMCARecord, "mca30")
# mca31 = Cpt(EpicsMCARecord, "mca31")
# mca32 = Cpt(EpicsMCARecord, "mca32")
current_channel = Cpt(EpicsSignalRO, "CurrentChannel", auto_monitor=True)
num_lines = Cpt(
bec_utils.ConfigSignal,
name="num_lines",
kind="config",
config_storage_name="mcs_config",
)
def __init__(
self,
prefix="",
*,
name,
kind=None,
read_attrs=None,
configuration_attrs=None,
parent=None,
device_manager=None,
sim_mode=False,
mcs_config=None,
**kwargs,
):
self.mcs_config = {
f"{name}_num_lines": 1,
}
if mcs_config is not None:
[self.mcs_config.update({f"{name}_{key}": value}) for key, value in mcs_config.items()]
super().__init__(
prefix=prefix,
name=name,
kind=kind,
read_attrs=read_attrs,
configuration_attrs=configuration_attrs,
parent=parent,
**kwargs,
)
if device_manager is None and not sim_mode:
raise MCSError("Add DeviceManager to initialization or init with sim_mode=True")
self.name = name
self._stream_ttl = 1800
self.wait_for_connection() # Make sure to be connected before talking to PVs
if not sim_mode:
self.device_manager = device_manager
self._producer = self.device_manager.producer
else:
self._producer = bec_utils.MockProducer()
self.device_manager = bec_utils.MockDeviceManager()
# TODO mack mock connector class
# self._consumer = self.device_manager.connector.consumer
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
# TODO
self.scaninfo.username = "e21206"
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"}
self.filewriter = FileWriterMixin(self.service_cfg)
self._stopped = False
self._acquisition_done = False
self._lock = threading.RLock()
self.counter = 0
self.n_points = 0
self._init_mcs()
def _init_mcs(self) -> None:
"""Init parameters for mcs card 9m
channel_advance: 0/1 -> internal / external
channel1_source: 0/1 -> int clock / external source
user_led: 0/1 -> off/on
max_output : num of channels 0...32, uncomment top for more than 5
input_mode: operation mode -> Mode 3 for external trigger, check manual for more info
input_polarity: triggered between falling and falling edge -> use inverted signal from ddg
"""
self.channel_advance.set(ChannelAdvance.EXTERNAL)
self.channel1_source.set(ChannelAdvance.INTERNAL)
self.user_led.set(0)
self.mux_output.set(5)
self._set_trigger(TriggerSource.MODE3)
self.input_polarity.set(0)
self.output_polarity.set(1)
self.count_on_start.set(0)
self.mca_names = [signal for signal in self.component_names if signal.startswith("mca")]
self.mca_data = defaultdict(lambda: [])
for mca in self.mca_names:
signal = getattr(self, mca)
signal.subscribe(self._on_mca_data, run=False)
self.current_channel.subscribe(self._progress_update, run=False)
def _progress_update(self, value, **kwargs) -> None:
num_lines = self.num_lines.get()
max_value = self.scaninfo.num_points
self._run_subs(
sub_type=self.SUB_PROGRESS,
value=self.counter * int(self.scaninfo.num_points / num_lines) + max(value - 1, 0),
max_value=max_value,
done=bool(max_value == self.counter),
)
@threadlocked
def _on_mca_data(self, *args, obj=None, **kwargs) -> None:
if not isinstance(kwargs["value"], (list, np.ndarray)):
return
self.mca_data[obj.attr_name] = kwargs["value"][1:]
if len(self.mca_names) != len(self.mca_data):
return
# ref_entry = self.mca_data[self.mca_names[0]]
# if not ref_entry:
# self.mca_data = defaultdict(lambda: [])
# return
# if isinstance(ref_entry, list) and (ref_entry > 0):
# return
self._updated = True
self.counter += 1
if (self.scaninfo.scan_type == "fly" and self.counter == self.num_lines.get()) or (
self.scaninfo.scan_type == "step" and self.counter == self.scaninfo.num_points
):
self._acquisition_done = True
self.stop_all.put(1, use_complete=False)
self._send_data_to_bec()
self.erase_all.put(1)
# Require wait for
# time.sleep(0.01)
self.mca_data = defaultdict(lambda: [])
self.counter = 0
return
self.erase_start.set(1)
self._send_data_to_bec()
self.mca_data = defaultdict(lambda: [])
def _send_data_to_bec(self) -> None:
if self.scaninfo.scan_msg is None:
return
metadata = self.scaninfo.scan_msg.metadata
metadata.update(
{
"async_update": "append",
"num_lines": self.num_lines.get(),
}
)
msg = BECMessage.DeviceMessage(
signals=dict(self.mca_data),
metadata=self.scaninfo.scan_msg.metadata,
).dumps()
self._producer.xadd(
topic=MessageEndpoints.device_async_readback(
scanID=self.scaninfo.scanID, device=self.name
),
msg={"data": msg},
expire=self._stream_ttl,
)
def _prep_det(self) -> None:
self._set_acquisition_params()
self._set_trigger(TriggerSource.MODE3)
def _set_acquisition_params(self) -> None:
if self.scaninfo.scan_type == "step":
self.n_points = int(self.scaninfo.frames_per_trigger + 1)
elif self.scaninfo.scan_type == "fly":
self.n_points = int(self.scaninfo.num_points / int(self.num_lines.get()) + 1)
else:
raise MCSError(f"Scantype {self.scaninfo} not implemented for MCS card")
if self.n_points > 10000:
raise MCSError(
f"Requested number of points N={self.n_points} exceeds hardware limit of mcs card 10000 (N-1)"
)
self.num_use_all.set(self.n_points)
self.preset_real.set(0)
def _set_trigger(self, trigger_source: TriggerSource) -> None:
"""7 Modes, see TriggerSource
Mode3 for cSAXS"""
value = int(trigger_source)
self.input_mode.set(value)
def _prep_readout(self) -> None:
"""Set readout mode of mcs card
Check ReadoutMode class for more information about options
"""
# self.read_mode.set(ReadoutMode.EVENT)
self.erase_all.put(1)
self.read_mode.set(ReadoutMode.EVENT)
def _force_readout_mcs_card(self) -> None:
self.read_all.put(1, use_complete=False)
def stage(self) -> List[object]:
"""stage the detector and file writer"""
logger.info("Stage mcs")
self.scaninfo.load_scan_metadata()
self._prep_det()
self._prep_readout()
# msg = BECMessage.FileMessage(file_path=self.filepath, done=False)
# self._producer.set_and_publish(
# MessageEndpoints.public_file(self.scaninfo.scanID, "mcs_csaxs"),
# msg.dumps(),
# )
self.arm_acquisition()
logger.info("Waiting for mcs to be armed")
while True:
det_ctrl = self.acquiring.read()[self.acquiring.name]["value"]
if det_ctrl == 1:
break
time.sleep(0.005)
logger.info("mcs is ready and running")
# time.sleep(5)
return super().stage()
def unstage(self) -> List[object]:
"""unstage"""
logger.info("Waiting for mcs to finish acquisition")
while not self._acquisition_done:
# monitor signal instead?
if self._stopped:
break
time.sleep(0.005)
self._acquisition_done = False
self._stopped = False
logger.info("mcs done")
return super().unstage()
def arm_acquisition(self) -> None:
"""Arm acquisition
Options:
Start: start_all
Erase/Start: erase_start
"""
self.counter = 0
self.erase_start.set(1)
# self.start_all.set(1)
def stop(self, *, success=False) -> None:
"""Stop acquisition
Stop or Stop and Erase
"""
self.stop_all.set(1)
# self.erase_all.set(1)
self._stopped = True
self._acquisition_done = True
self.counter = 0
super().stop(success=success)
# Automatically connect to test environmenr if directly invoked
if __name__ == "__main__":
mcs = McsCsaxs(name="mcs", prefix="X12SA-MCS:", sim_mode=True)
mcs.stage()
mcs.unstage()

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

@ -1,32 +1,117 @@
import enum
import json
import os
import time
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 ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV
from ophyd import DetectorBase, Device
from ophyd import ADComponent as ADCpt
from ophyd_devices.utils import bec_utils as bec_utils
from bec_lib.core import BECMessage, MessageEndpoints
from bec_lib.core.file_utils import FileWriterMixin
from bec_lib.core import bec_logger
from ophyd_devices.epics.devices.bec_scaninfo_mixin import BecScaninfoMixin
logger = bec_logger.logger
class PilatusDetectorCamEx(PilatusDetectorCam, FileBase):
class PilatusError(Exception):
pass
class TriggerSource(int, enum.Enum):
INTERNAL = 0
EXT_ENABLE = 1
EXT_TRIGGER = 2
MULTI_TRIGGER = 3
ALGINMENT = 4
class SlsDetectorCam(Device): # CamBase, FileBase):
# detector_type = ADCpt(EpicsSignalRO, "DetectorType_RBV")
# setting = ADCpt(EpicsSignalWithRBV, "Setting")
# beam_energy = ADCpt(EpicsSignalWithRBV, "BeamEnergy")
# enable_trimbits = ADCpt(EpicsSignalWithRBV, "Trimbits")
# bit_depth = ADCpt(EpicsSignalWithRBV, "BitDepth")
# trigger_software = ADCpt(EpicsSignal, "TriggerSoftware")
# high_voltage = ADCpt(EpicsSignalWithRBV, "HighVoltage")
# Receiver and data callback
# receiver_mode = ADCpt(EpicsSignalWithRBV, "ReceiverMode")
# receiver_stream = ADCpt(EpicsSignalWithRBV, "ReceiverStream")
# enable_data = ADCpt(EpicsSignalWithRBV, "UseDataCallback")
# missed_packets = ADCpt(EpicsSignalRO, "ReceiverMissedPackets_RBV")
# # Direct settings access
# setup_file = ADCpt(EpicsSignal, "SetupFile")
# load_setup = ADCpt(EpicsSignal, "LoadSetup")
# command = ADCpt(EpicsSignal, "Command")
# Mythen 3
# counter_mask = ADCpt(EpicsSignalWithRBV, "CounterMask")
# counter1_threshold = ADCpt(EpicsSignalWithRBV, "Counter1Threshold")
# counter2_threshold = ADCpt(EpicsSignalWithRBV, "Counter2Threshold")
# counter3_threshold = ADCpt(EpicsSignalWithRBV, "Counter3Threshold")
# gate1_delay = ADCpt(EpicsSignalWithRBV, "Gate1Delay")
# gate1_width = ADCpt(EpicsSignalWithRBV, "Gate1Width")
# gate2_delay = ADCpt(EpicsSignalWithRBV, "Gate2Delay")
# gate2_width = ADCpt(EpicsSignalWithRBV, "Gate2Width")
# gate3_delay = ADCpt(EpicsSignalWithRBV, "Gate3Delay")
# gate3_width = ADCpt(EpicsSignalWithRBV, "Gate3Width")
# Moench
# json_frame_mode = ADCpt(EpicsSignalWithRBV, "JsonFrameMode")
# json_detector_mode = ADCpt(EpicsSignalWithRBV, "JsonDetectorMode")
# Eiger9M
# delay_time = ADCpt(EpicsSignalWithRBV, "DelayTime")
# num_frames = ADCpt(EpicsSignalWithRBV, "NumFrames")
# acquire = ADCpt(EpicsSignal, "Acquire")
# acquire_time = ADCpt(EpicsSignal, 'AcquireTime')
# detector_state = ADCpt(EpicsSignalRO, "DetectorState_RBV")
# threshold_energy = ADCpt(EpicsSignalWithRBV, "ThresholdEnergy")
# num_gates = ADCpt(EpicsSignalWithRBV, "NumGates")
# num_cycles = ADCpt(EpicsSignalWithRBV, "NumCycles")
# timing_mode = ADCpt(EpicsSignalWithRBV, "TimingMode")
# Pilatus_2 300k
num_images = ADCpt(EpicsSignalWithRBV, "NumImages")
num_exposures = ADCpt(EpicsSignalWithRBV, "NumExposures")
delay_time = ADCpt(EpicsSignalWithRBV, "NumExposures")
trigger_mode = ADCpt(EpicsSignalWithRBV, "TriggerMode")
acquire = ADCpt(EpicsSignal, "Acquire")
armed = ADCpt(EpicsSignalRO, "Armed")
read_file_timeout = ADCpt(EpicsSignal, "ImageFileTmot")
detector_state = ADCpt(EpicsSignalRO, "StatusMessage_RBV")
status_message_camserver = ADCpt(EpicsSignalRO, "StringFromServer_RBV")
acquire_time = ADCpt(EpicsSignal, "AcquireTime")
acquire_period = ADCpt(EpicsSignal, "AcquirePeriod")
threshold_energy = ADCpt(EpicsSignalWithRBV, "ThresholdEnergy")
file_path = ADCpt(EpicsSignalWithRBV, "FilePath")
file_name = ADCpt(EpicsSignalWithRBV, "FileName")
file_number = ADCpt(EpicsSignalWithRBV, "FileNumber")
auto_increment = ADCpt(EpicsSignalWithRBV, "AutoIncrement")
file_template = ADCpt(EpicsSignalWithRBV, "FileTemplate")
file_format = ADCpt(EpicsSignalWithRBV, "FileNumber")
gap_fill = ADCpt(EpicsSignalWithRBV, "GapFill")
class PilatusCsaxs(DetectorBase):
"""Pilatus_2 300k detector for CSAXS
Parent class: DetectorBase
Device class: PilatusDetectorCamEx
Attributes:
name str: 'pilatus_2'
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:")
cam = ADCpt(SlsDetectorCam, "cam1:")
def __init__(
self,
@ -38,11 +123,9 @@ class PilatusCsaxs(DetectorBase):
configuration_attrs=None,
parent=None,
device_manager=None,
sim_mode=False,
**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,
@ -52,49 +135,92 @@ 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/"}
if device_manager is None and not sim_mode:
raise PilatusError("Add DeviceManager to initialization or init with sim_mode=True")
self.name = name
self.wait_for_connection() # Make sure to be connected before talking to PVs
if not sim_mode:
from bec_lib.core.bec_service import SERVICE_CONFIG
self.device_manager = device_manager
self._producer = self.device_manager.producer
self.service_cfg = SERVICE_CONFIG.config["service_config"]["file_writer"]
else:
self._producer = bec_utils.MockProducer()
self.device_manager = bec_utils.MockDeviceManager()
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self.scaninfo.load_scan_metadata()
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.scaninfo.username}/Data10/"}
self.scaninfo = BecScaninfoMixin(device_manager, sim_mode)
self.filepath_h5 = ""
self.filewriter = FileWriterMixin(self.service_cfg)
self.num_frames = 0
self.readout = 0.003 # 3 ms
self.triggermode = 0 # 0 : internal, scan must set this if hardware triggered
self.readout = 1e-3 # 3 ms
def _get_current_scan_msg(self) -> BECMessage.ScanStatusMessage:
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 _prep_det(self) -> None:
# TODO slow reaction, seemed to have timeout.
self._set_det_threshold()
self._set_acquisition_params()
# set pilatus threshol
self._set_threshold()
def _set_det_threshold(self) -> None:
# threshold_energy PV exists on Eiger 9M?
factor = 1
if self.cam.threshold_energy._metadata["units"] == "eV":
factor = 1000
setp_energy = int(self.mokev * factor)
threshold = self.cam.threshold_energy.read()[self.cam.threshold_energy.name]["value"]
if not np.isclose(setp_energy / 2, threshold, rtol=0.05):
self.cam.threshold_energy.set(setp_energy / 2)
# set Epic PVs for filewriting
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
self.cam.file_number.set(0) # first iter
self.cam.file_format.set(0) # 0: TIFF
self.cam.file_template.set("%s%s_%5.5d.cbf")
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(int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger))
self.cam.num_exposures.set(1)
self._set_trigger(TriggerSource.EXT_ENABLE) # EXT_TRIGGER)
# compile zmq stream for data transfer
scan_dir = self.filewriter._get_scan_directory(
scan_bundle=1000, scan_number=self.scan_number, leading_zeros=5
def _set_trigger(self, trigger_source: TriggerSource) -> None:
"""Set trigger source for the detector, either directly to value or TriggerSource.* with
INTERNAL = 0
EXT_ENABLE = 1
EXT_TRIGGER = 2
MULTI_TRIGGER = 3
ALGINMENT = 4
"""
value = int(trigger_source)
self.cam.trigger_mode.set(value)
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.filepath_h5 = self.filewriter.compile_full_filename(
self.scaninfo.scan_number, "pilatus_2.h5", 1000, 5, True
)
self.cam.file_path.put(f"/dev/shm/zmq/")
self.cam.file_name.put(f"{self.scaninfo.username}_2_{self.scaninfo.scan_number:05d}")
self.cam.auto_increment.put(1) # auto increment
self.cam.file_number.put(0) # first iter
self.cam.file_format.put(0) # 0: TIFF
self.cam.file_template.put("%s%s_%5.5d.cbf")
# compile filename
basepath = f"/sls/X12SA/data/{self.scaninfo.username}/Data10/pilatus_2/"
self.destination_path = os.path.join(
self.service_cfg["base_path"]
) # os.path.join(self.service_cfg["base_path"], scan_dir)
basepath,
self.filewriter.get_scan_directory(self.scaninfo.scan_number, 1000, 5),
)
# Make directory if needed
os.makedirs(os.path.dirname(self.destination_path), exist_ok=True)
data_msg = {
"source": [
{
@ -104,6 +230,7 @@ class PilatusCsaxs(DetectorBase):
}
]
}
logger.info(data_msg)
headers = {"Content-Type": "application/json", "Accept": "application/json"}
@ -112,6 +239,7 @@ class PilatusCsaxs(DetectorBase):
data=json.dumps(data_msg),
headers=headers,
)
logger.info(f"{res.status_code} - {res.text} - {res.content}")
if not res.ok:
res.raise_for_status()
@ -119,14 +247,14 @@ class PilatusCsaxs(DetectorBase):
# prepare writer
data_msg = [
"zmqWriter",
self.username,
self.scaninfo.username,
{
"addr": "tcp://x12sa-pd-2:8888",
"dst": ["file"],
"numFrm": self.num_frames,
"numFrm": int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger),
"timeout": 2000,
"ifType": "PULL",
"user": self.username,
"user": self.scaninfo.username,
},
]
@ -136,82 +264,123 @@ class PilatusCsaxs(DetectorBase):
headers=headers,
)
logger.info(f"{res.status_code} - {res.text} - {res.content}")
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,
# Wait for server to become available again
time.sleep(0.1)
headers = {"Content-Type": "application/json", "Accept": "application/json"}
data_msg = [
"zmqWriter",
self.scaninfo.username,
{
"frmCnt": int(self.scaninfo.num_points * self.scaninfo.frames_per_trigger),
"timeout": 2000,
},
]
logger.info(f"{res.status_code} -{res.text} - {res.content}")
try:
res = requests.put(
url="http://xbl-daq-34:8091/pilatus_2/wait",
data=json.dumps(data_msg),
# headers=headers,
)
logger.info(f"{res}")
if not res.ok:
res.raise_for_status()
except Exception as exc:
logger.info(f"Pilatus2 wait threw Exception: {exc}")
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
"""
try:
res = requests.delete(url="http://x12sa-pd-2:8080/stream/pilatus_2")
if not res.ok:
res.raise_for_status()
except Exception as exc:
logger.info(f"Pilatus2 delete threw Exception: {exc}")
def _stop_file_writer(self) -> None:
res = requests.put(
url="http://xbl-daq-34:8091/pilatus_2/stop",
# data=json.dumps(data_msg),
# headers=headers,
)
if not res.ok:
res.raise_for_status()
def stage(self) -> List[object]:
"""stage the detector and file writer"""
self._close_file_writer()
self._stop_file_writer()
self.scaninfo.load_scan_metadata()
self.mokev = self.device_manager.devices.mokev.obj.read()[
self.device_manager.devices.mokev.name
]["value"]
logger.info("Waiting for pilatus2 to be armed")
self._prep_det()
logger.info("Pilatus2 armed")
logger.info("Waiting for pilatus2 zmq stream to be ready")
self._prep_file_writer()
logger.info("Pilatus2 zmq ready")
msg = BECMessage.FileMessage(
file_path=self.filepath_h5, done=False, metadata={"input_path": self.destination_path}
)
return super().stage()
def pre_scan(self) -> None:
self.acquire()
def unstage(self) -> List[object]:
headers = {"Content-Type": "application/json", "Accept": "application/json"}
data_msg = [
"zmqWriter",
self.username,
{
"frmCnt": self.num_frames,
"timeout": 2000,
"ifType": "PULL",
"user": self.username,
},
]
logger.info(data_msg)
res = requests.put(
url="http://xbl-daq-34:8091/pilatus_1/run",
data=json.dumps(data_msg),
headers=headers,
)
# Reset triggermode to internal
"""unstage the detector and file writer"""
# Reset to software trigger
self.triggermode = 0
if not res.ok:
res.raise_for_status()
# TODO if images are missing, consider adding delay
self._close_file_writer()
self._stop_file_writer()
# Only sent this out once data is written to disk since cbf to hdf5 converter will be triggered
msg = BECMessage.FileMessage(
file_path=self.filepath_h5, done=True, metadata={"input_path": self.destination_path}
)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scaninfo.scanID, self.name),
msg.dumps(),
)
self._producer.set_and_publish(
MessageEndpoints.file_event(self.name),
msg.dumps(),
)
logger.info("Pilatus2 done")
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._stop_file_writer()
# self.unstage()
super().stop(success=success)
self._stopped = True
# Automatically connect to test environmenr if directly invoked
if __name__ == "__main__":
pilatus_2 = PilatusCsaxs(name="pilatus_2", prefix="X12SA-ES-PILATUS300K:", sim_mode=True)
pilatus_2.stage()

2
ophyd_devices/epics/devices/specMotors.py
View File

@ -181,7 +181,7 @@ class MonoTheta2(VirtualEpicsSignalRO):
MONO_THETA2_OFFSETS_FILENAME = (
"/import/work/sls/spec/local/X12SA/macros/spec_data/mono_th2_offsets.txt"
"/sls/X12SA/data/gac-x12saop/spec/macros/spec_data/mono_th2_offsets.txt"
)

50
ophyd_devices/galil/sgalil_ophyd.py
View File

@ -49,6 +49,9 @@ class GalilController(Controller):
"galil_show_all",
"socket_put_and_receive",
"socket_put_confirmed",
"sgalil_reference",
"fly_grid_scan",
"read_encoder_position",
]
def __init__(
@ -150,7 +153,10 @@ class GalilController(Controller):
return var
def stop_all_axes(self) -> str:
return self.socket_put_and_receive(f"XQ#STOP,1")
# return self.socket_put_and_receive(f"XQ#STOP,1")
# Command stops all threads and motors!
# self.socket_put_and_receive(f"ST")
return self.socket_put_and_receive(f"AB")
def axis_is_referenced(self) -> bool:
return bool(float(self.socket_put_and_receive(f"MG allaxref").strip()))
@ -244,7 +250,8 @@ class GalilController(Controller):
end_x: float,
interval_x: int,
exp_time: float,
readtime: float,
readout_time: float,
**kwargs,
) -> tuple:
"""_summary_
@ -256,7 +263,7 @@ class GalilController(Controller):
end_x (float): end position of x axis (slow axis)
interval_x (int): number of points in x axis
exp_time (float): exposure time in seconds
readtime (float): readout time in seconds, minimum of .5e-3s (0.5ms)
readout_time (float): readout time in seconds, minimum of .5e-3s (0.5ms)
Raises:
@ -264,16 +271,24 @@ class GalilController(Controller):
LimitError: Raised if the speed is above 2mm/s or below 0.02mm/s
"""
# time.sleep(0.2)
#
axes_referenced = self.axis_is_referenced()
sign_y = self._axis[ord("c") - 97].sign
sign_x = self._axis[ord("e") - 97].sign
# Check limits
# TODO check sign of stage, or not necessary
check_values = [start_y, end_y, start_x, end_x]
for val in check_values:
self.check_value(val)
speed = np.abs(end_y - start_y) / ((interval_y) * exp_time + (interval_y - 1) * readtime)
start_x *= sign_x
end_x *= sign_x
start_y *= sign_y
end_y *= sign_y
speed = np.abs(end_y - start_y) / (
(interval_y) * exp_time + (interval_y - 1) * readout_time
)
if speed > 2.00 or speed < 0.02:
raise LimitError(
f"Speed of {speed:.03f}mm/s is outside of acceptable range of 0.02 to 2 mm/s"
@ -284,7 +299,7 @@ class GalilController(Controller):
n_samples = int(interval_y * interval_x)
# Hard coded to maximum offset of 0.1mm to avoid long motions.
self.socket_put_and_receive(f"off={(0*0.1/2*1000):f}")
self.socket_put_and_receive(f"off={(0):f}")
self.socket_put_and_receive(f"a_start={start_y:.04f};a_end={end_y:.04f};speed={speed:.04f}")
self.socket_put_and_receive(
f"b_start={start_x:.04f};gridmax={gridmax:d};b_step={step_grid:.04f}"
@ -298,6 +313,7 @@ class GalilController(Controller):
# threading.Thread(target=_while_in_motion(3, n_samples), daemon=True).start()
# self._while_in_motion(3, n_samples)
# TODO this is for reading out positions, readout is limited by stage triggering
def _while_in_motion(self, thread_id: int, n_samples: int) -> tuple:
last_readout = 0
val_axis2 = [] # y axis
@ -390,7 +406,7 @@ class GalilSetpointSignal(GalilSignalBase):
Returns:
float: setpoint / target value
"""
return self.setpoint
return self.setpoint * self.parent.sign
@retry_once
@threadlocked
@ -664,6 +680,22 @@ class SGalilMotor(Device, PositionerBase):
self.controller.stop_all_axes()
return super().stop(success=success)
def kickoff(
self,
metadata: dict,
**kwargs,
) -> None:
self.controller.fly_grid_scan(
kwargs.get("start_y"),
kwargs.get("end_y"),
kwargs.get("interval_y"),
kwargs.get("start_x"),
kwargs.get("end_x"),
kwargs.get("interval_x"),
kwargs.get("exp_time"),
kwargs.get("readout_time"),
)
if __name__ == "__main__":
mock = False

144
ophyd_devices/utils/bec_utils.py Normal file
View File

@ -0,0 +1,144 @@
import time
from bec_lib.core import bec_logger
from ophyd import Signal, Kind
from ophyd_devices.utils.socket import data_shape, data_type
logger = bec_logger.logger
DEFAULT_EPICSSIGNAL_VALUE = object()
class MockProducer:
def set_and_publish(self, endpoint: str, msgdump: str):
logger.info(f"BECMessage to {endpoint} with msg dump {msgdump}")
class MockDeviceManager:
def __init__(self) -> None:
self.devices = devices()
class OphydObject:
def __init__(self) -> None:
self.name = "mock_mokev"
self.obj = mokev()
class devices:
def __init__(self):
self.mokev = OphydObject()
class mokev:
def __init__(self):
self.name = "mock_mokev"
def read(self):
return {self.name: {"value": 16.0, "timestamp": time.time()}}
class ConfigSignal(Signal):
def __init__(
self,
*,
name,
value=0,
timestamp=None,
parent=None,
labels=None,
kind=Kind.hinted,
tolerance=None,
rtolerance=None,
metadata=None,
cl=None,
attr_name="",
config_storage_name: str = "config_storage",
):
super().__init__(
name=name,
value=value,
timestamp=timestamp,
parent=parent,
labels=labels,
kind=kind,
tolerance=tolerance,
rtolerance=rtolerance,
metadata=metadata,
cl=cl,
attr_name=attr_name,
)
self.storage_name = config_storage_name
def get(self):
self._readback = getattr(self.parent, self.storage_name)[self.name]
return self._readback
def put(
self,
value,
connection_timeout=1,
callback=None,
timeout=1,
**kwargs,
):
"""Using channel access, set the write PV to `value`.
Keyword arguments are passed on to callbacks
Parameters
----------
value : any
The value to set
connection_timeout : float, optional
If not already connected, allow up to `connection_timeout` seconds
for the connection to complete.
use_complete : bool, optional
Override put completion settings
callback : callable
Callback for when the put has completed
timeout : float, optional
Timeout before assuming that put has failed. (Only relevant if
put completion is used.)
"""
old_value = self.get()
timestamp = time.time()
getattr(self.parent, self.storage_name)[self.name] = value
super().put(value, timestamp=timestamp, force=True)
self._run_subs(
sub_type=self.SUB_VALUE,
old_value=old_value,
value=value,
timestamp=timestamp,
)
def describe(self):
"""Provide schema and meta-data for :meth:`~BlueskyInterface.read`
This keys in the `OrderedDict` this method returns must match the
keys in the `OrderedDict` return by :meth:`~BlueskyInterface.read`.
This provides schema related information, (ex shape, dtype), the
source (ex PV name), and if available, units, limits, precision etc.
Returns
-------
data_keys : OrderedDict
The keys must be strings and the values must be dict-like
with the ``event_model.event_descriptor.data_key`` schema.
"""
if self._readback is DEFAULT_EPICSSIGNAL_VALUE:
val = self.get()
else:
val = self._readback
return {
self.name: {
"source": f"{self.parent.prefix}:{self.name}",
"dtype": data_type(val),
"shape": data_shape(val),
}
}

1
setup.py
View File

@ -11,6 +11,7 @@ if __name__ == "__main__":
"bec_lib",
"numpy",
"pyyaml",
"std_daq_client",
"pyepics",
],
extras_require={"dev": ["pytest", "pytest-random-order", "black"]},