302 lines
13 KiB
Python
302 lines
13 KiB
Python
"""Module to automatically set the gains for the selected amplifiers"""
|
|
|
|
import builtins
|
|
import time
|
|
from bisect import bisect_right
|
|
|
|
import xraydb
|
|
from bec_lib import bec_logger
|
|
from ophyd_devices import CompareStatus
|
|
|
|
from ...devices.absorber import STATUS as ABS_STATUS
|
|
from ...devices.eh_shutter import STATUS as EH_PH_STATUS
|
|
from ...devices.ionization_chambers.ionization_chamber_enums import AmplifierEnable
|
|
from ...devices.nidaq.nidaq_enums import EpicsMode, NidaqState
|
|
from ...devices.op_shutter import STATUS as OP_PH_STATUS
|
|
|
|
logger = bec_logger.logger
|
|
|
|
EMIN = -100
|
|
EMAX = 200
|
|
|
|
MIN_RING_CURRENT = 5 # Minimum ring current to use auto-gain
|
|
NOMINAL_RING_CURRENT = 400 # Nominal ring current of SLS2
|
|
|
|
TIMEOUT_PUT_PV = 5 # Timeout to set a PV
|
|
|
|
MONO_VELOCITY = 20 # Move velocity in deg/s
|
|
TIMEOUT_MONO_MOVE = 30 # Timeout to finish a movement on the mono
|
|
|
|
LINKED_CHANNELS = {"ic0", "ic1"}
|
|
AVAILABLE_GAINS = [1e6, 1e7, 5e7, 1e8, 1e9] # ascending order
|
|
MIN_SIGNAL = 0.05 # Minimum signal to count as valid signal
|
|
FULL_SCALE_V = 10.0 # NIDAQ AI full-scale range
|
|
SAFETY_MARGIN = 0.9 # keep max signal under 90% of full scale
|
|
|
|
|
|
class AutoGainError(Exception):
|
|
"""AutoGain specific error"""
|
|
|
|
|
|
class AutoGain:
|
|
"""Module to automatically set the gains for the selected amplifiers"""
|
|
|
|
def __init__(self):
|
|
dev = builtins.__dict__.get("dev")
|
|
bec = builtins.__dict__.get("bec")
|
|
if dev is None:
|
|
raise AutoGainError("Did not get dev")
|
|
if bec is None:
|
|
raise AutoGainError("Did not get bec")
|
|
self.dev = dev
|
|
self.bec = bec
|
|
|
|
def start(
|
|
self,
|
|
element: str,
|
|
edge: str,
|
|
amplifier: list[str] | None = None,
|
|
comp_ring_current: bool = True,
|
|
) -> None:
|
|
"""Start the auto-gain sequence. Measure the signals of the specified
|
|
amplifiers and set the gains accordingly. Makes sure there is actually beam available.
|
|
|
|
Note:
|
|
If 'ic0' and 'ic1' are in the amplifier argument, the same gain will be assigned
|
|
to both amplifiers to get the best glitch removal during processing later.
|
|
|
|
Args:
|
|
element(str): Element which defines the energy at which the gain will be set, e.g. 'Cu'
|
|
edge(str): Corresponding edge, e.g. 'L1'
|
|
amplifier(list[str]): Amplifiers where auto-gain should be applied to
|
|
Defaults to all amplifiers -> ['ic0', 'ic1', 'ic2', 'pips']
|
|
comp_ring_current(bool): Respects the current ring current and calculates the gain(s)
|
|
for a nominal ring current of 400 mA. Defaults to True
|
|
|
|
Raises:
|
|
If NIDAQ is not in measurement mode
|
|
If Ring current is below 5 mA
|
|
If Absorber, OP Photon Shutter or EH Photon Shutter is closed
|
|
If a bec scan is running
|
|
If the energy for the supplied element/edge cannot be found
|
|
If the energy is outside the movement range of the monochromator
|
|
If a selected amplifier is switched off
|
|
If the high voltage of a selected ionization chamber is not enabled or < 1000 V
|
|
If the gas filling of a selected ionization chamber is not OK
|
|
If the ring current drops to 0 mA during the measurement (beamdump)
|
|
"""
|
|
|
|
if amplifier is None:
|
|
amplifier = ["ic0", "ic1", "ic2", "pips"]
|
|
|
|
# Make sure NIDAQ is in standby mode
|
|
if self.dev.nidaq.state.get() != NidaqState.STANDBY:
|
|
raise AutoGainError("NIDAQ was not in Standby mode, cannot proceed.")
|
|
|
|
# Check for beam availability
|
|
if self._get_ring_current() < MIN_RING_CURRENT:
|
|
raise AutoGainError(f"Ring current is below {MIN_RING_CURRENT} mA")
|
|
if self.dev.abs.status.get() != ABS_STATUS.OPEN:
|
|
raise AutoGainError("Absorber is closed, no beam")
|
|
if self.dev.op_sh.status.get() != OP_PH_STATUS.NOT_CLOSED:
|
|
raise AutoGainError("OP Photon Shutter is closed, no beam")
|
|
if self.dev.eh_sh.status.get() != EH_PH_STATUS.NOT_CLOSED:
|
|
raise AutoGainError("EH Photon Shutter is closed, no beam")
|
|
|
|
# Check if no scan is running
|
|
scan_id = self.bec.queue.scan_storage.current_scan_id
|
|
if len(scan_id) > 0:
|
|
raise AutoGainError(f"Scan with ID {scan_id} is currently running, cannot continue")
|
|
|
|
# Get edge energy
|
|
energy = xraydb.xray_edge(element, edge, True)
|
|
if energy is None:
|
|
raise ValueError(f"Could not find edge energy for element/edge {element}/{edge}")
|
|
emin = energy + EMIN
|
|
emax = energy + EMAX
|
|
|
|
# Check range of mono
|
|
low_limit = self.dev.mo1_bragg.low_lim.get()
|
|
high_limit = self.dev.mo1_bragg.high_lim.get()
|
|
if emin < low_limit or emax > high_limit:
|
|
raise ValueError(
|
|
f"Chosen element/edge {element}/{edge} with edge energy of {energy}"
|
|
+ " is outside of accessible range of monochromator "
|
|
+ f"{low_limit:.1f} eV - {high_limit:.1f} eV"
|
|
)
|
|
|
|
# Map amplifier names to their NIDAQ channels
|
|
channel_map = {
|
|
"ic0": {"signal": self.dev.nidaq.ai0, "self.dev": self.dev.ic0},
|
|
"ic1": {"signal": self.dev.nidaq.ai2, "self.dev": self.dev.ic1},
|
|
"ic2": {"signal": self.dev.nidaq.ai4, "self.dev": self.dev.ic2},
|
|
"pips": {"signal": self.dev.nidaq.ai6, "self.dev": self.dev.pips},
|
|
}
|
|
active_channels = {name: ch for name, ch in channel_map.items() if name in amplifier}
|
|
|
|
# Check if amplifieres are switched on
|
|
for name, ch in active_channels.items():
|
|
if ch["self.dev"].amp.cOnOff.get() != AmplifierEnable.ON:
|
|
raise AutoGainError(f"Amplifier of self.device {name} is not enabled")
|
|
|
|
# Check high voltage on ionization chambers
|
|
for name, ch in active_channels.items():
|
|
if name != "pips":
|
|
if ch["self.dev"].hv_en.ena.get() != 1:
|
|
raise AutoGainError(f"High voltage of ionization chamber {name} is not enabled")
|
|
if ch["self.dev"].hv.hv_v.get() < 1000:
|
|
raise AutoGainError(f"HV voltage of ionization chamber {name} is < 1000")
|
|
if ch["self.dev"].hv.grid_v.get() < 1000:
|
|
raise AutoGainError(f"Grid voltage of ionization chamber {name} is < 1000")
|
|
|
|
# Check gas filling of ionization chambers
|
|
for name, ch in active_channels.items():
|
|
if name != "pips":
|
|
if ch["self.dev"].gmes.status.get() != 1:
|
|
raise AutoGainError(f"Gas filling of ionization chamber {name} is not OK")
|
|
|
|
logger.info("All checks done, start preparing for measurement")
|
|
|
|
# Get initial monochromator position and velocity
|
|
init_pos = self.dev.mo1_bragg.position.get()
|
|
init_vel = self.dev.mo1_bragg.velocity.get()
|
|
|
|
logger.info(f"Move mono to start of {emin} eV")
|
|
self.dev.mo1_bragg.velocity.put(MONO_VELOCITY)
|
|
status = CompareStatus(self.dev.mo1_bragg.velocity, MONO_VELOCITY)
|
|
status.wait(TIMEOUT_PUT_PV)
|
|
status = self.dev.mo1_bragg.move(emin)
|
|
status.wait(TIMEOUT_MONO_MOVE)
|
|
|
|
# Set NIDAQ to max mode
|
|
self.dev.nidaq.epics_mode.put(EpicsMode.MAX)
|
|
status = CompareStatus(self.dev.nidaq.epics_mode, EpicsMode.MAX)
|
|
status.wait(timeout=TIMEOUT_PUT_PV)
|
|
|
|
# Set gains to lowest gain
|
|
for name, ch in active_channels.items():
|
|
lowest_gain = AVAILABLE_GAINS[0]
|
|
ch["self.dev"].set_gain(lowest_gain)
|
|
ch["gain"] = lowest_gain
|
|
|
|
remeasure = True
|
|
logger.info(f"Start measurement from {emin} eV to {emax} eV")
|
|
while remeasure:
|
|
# Create temporary storage for max signal per channel
|
|
data = {name: 0 for name in active_channels}
|
|
|
|
# Measure current ring current
|
|
ring_current_1 = self._get_ring_current()
|
|
logger.info(f"Ring current right before measurement: {ring_current_1} mA")
|
|
if ring_current_1 == 0:
|
|
raise AutoGainError("Ring current dropped to 0 mA right before measurement")
|
|
|
|
# Scan range, recording the peak NIDAQ signal per channel
|
|
self.dev.mo1_bragg.move(emax).wait(timeout=TIMEOUT_MONO_MOVE)
|
|
status.wait(TIMEOUT_MONO_MOVE)
|
|
for name, ch in active_channels.items():
|
|
data[name] = max(data[name], ch["signal"].get())
|
|
|
|
# Rest max values of NIDAQ signals
|
|
self.dev.nidaq.epics_max_reset.put(True)
|
|
time.sleep(0.1)
|
|
|
|
# Measure current ring current again
|
|
ring_current_2 = self._get_ring_current()
|
|
logger.info(f"Ring current right after measurement: {ring_current_2} mA")
|
|
if ring_current_2 == 0:
|
|
raise AutoGainError("Ring current dropped to 0 mA during measurement")
|
|
ring_current = (ring_current_1 + ring_current_2) / 2
|
|
|
|
# Move back to first monochromator position
|
|
status = self.dev.mo1_bragg.move(emin)
|
|
|
|
# Choose gain per channel based on the max signal recorded during the scan
|
|
remeasure = False
|
|
processed = set()
|
|
|
|
for name, ch in active_channels.items():
|
|
if name in processed:
|
|
continue
|
|
|
|
# ic0/ic1 must always end up with the same gain -> treat whichever of
|
|
# them are present (one or both) as a single group for this decision
|
|
if name in LINKED_CHANNELS:
|
|
group_names = [n for n in LINKED_CHANNELS if n in active_channels]
|
|
else:
|
|
group_names = [name]
|
|
|
|
group_signals = []
|
|
for gname in group_names:
|
|
gsig = data[gname]
|
|
logger.info(f"Raw signal for device {gname} is {gsig} V")
|
|
if comp_ring_current:
|
|
gsig = gsig * NOMINAL_RING_CURRENT / ring_current
|
|
logger.info(f"Compensate for ring current, new raw signal is {gsig} V")
|
|
group_signals.append(gsig)
|
|
|
|
# worst case: the weakest signal drives "need more gain",
|
|
# the strongest signal drives "don't saturate"
|
|
raw_signal_min = min(group_signals)
|
|
raw_signal_max = max(group_signals)
|
|
gain_ref = ch[
|
|
"gain"
|
|
] # linked channels always share gain, so any member's value works
|
|
label = "/".join(group_names)
|
|
|
|
if raw_signal_min < MIN_SIGNAL:
|
|
logger.info(f"Raw signal for {label} is below {MIN_SIGNAL}")
|
|
if gain_ref == AVAILABLE_GAINS[-1]:
|
|
logger.warning(
|
|
f"Amplifier of {label} at highest gain {gain_ref} and still not"
|
|
+ f" measured signal above {MIN_SIGNAL}"
|
|
)
|
|
else:
|
|
next_gain = AVAILABLE_GAINS[
|
|
bisect_right(AVAILABLE_GAINS, gain_ref * 100) - 1
|
|
]
|
|
for gname in group_names:
|
|
active_channels[gname]["dev"].set_gain(next_gain)
|
|
active_channels[gname]["gain"] = next_gain
|
|
logger.info(f"Setting gain of {label} to {next_gain:.0e} and remeasure")
|
|
remeasure = True
|
|
else:
|
|
gain = max(
|
|
(
|
|
g
|
|
for g in AVAILABLE_GAINS
|
|
if raw_signal_max / gain_ref * g <= FULL_SCALE_V * SAFETY_MARGIN
|
|
),
|
|
default=min(AVAILABLE_GAINS),
|
|
)
|
|
for gname in group_names:
|
|
active_channels[gname]["dev"].set_gain(gain)
|
|
active_channels[gname]["gain"] = gain
|
|
logger.info(f"Calculated final gain for {label} of {gain:.0e}")
|
|
|
|
processed.update(group_names)
|
|
|
|
# Wait for mono to return to start position
|
|
status.wait(TIMEOUT_MONO_MOVE)
|
|
|
|
# Reset NIDAQ to mean mode
|
|
self.dev.nidaq.epics_mode.put(EpicsMode.MEAN)
|
|
status = CompareStatus(self.dev.nidaq.epics_mode, EpicsMode.MEAN)
|
|
status.wait(timeout=TIMEOUT_PUT_PV)
|
|
|
|
# Wait for mono to move to initial position and reset velocity
|
|
status = self.dev.mo1_bragg.move(init_pos)
|
|
status.wait(TIMEOUT_MONO_MOVE)
|
|
self.dev.mo1_bragg.velocity.put(init_vel)
|
|
status = CompareStatus(self.dev.mo1_bragg.velocity, init_vel)
|
|
status.wait(timeout=TIMEOUT_PUT_PV)
|
|
|
|
def _get_ring_current(self) -> float:
|
|
ring_current = 0
|
|
retries = 0
|
|
while ring_current == 0 and retries < 10:
|
|
ring_current = self.dev.curr.get()
|
|
retries += 1
|
|
time.sleep(0.01)
|
|
return ring_current
|