diff --git a/debye_bec/bec_ipython_client/plugins/auto_gain.py b/debye_bec/bec_ipython_client/plugins/auto_gain.py index 45fd085..bfc2957 100644 --- a/debye_bec/bec_ipython_client/plugins/auto_gain.py +++ b/debye_bec/bec_ipython_client/plugins/auto_gain.py @@ -6,7 +6,6 @@ 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 @@ -19,7 +18,7 @@ logger = bec_logger.logger EMIN = -100 EMAX = 200 -MIN_RING_CURRENT = 5 # Minimum ring current to use auto-gain +MIN_RING_CURRENT = -1 # 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 @@ -34,6 +33,22 @@ FULL_SCALE_V = 10.0 # NIDAQ AI full-scale range SAFETY_MARGIN = 0.9 # keep max signal under 90% of full scale +def wait_for_signal(signal, value, timeout: float) -> None: + """Function to wait for a signal to reach a certain value. + + Args: + signal: Signal to read from + value: Target value the signal should reach + timeout(float): Timeout in seconds before a TimeoutError is raised + + """ + start = time.time() + while time.time() - start < timeout: + if signal.get() == value: + return + raise TimeoutError(f"Signal {signal} did not reach value {value} within {timeout} seconds.") + + class AutoGainError(Exception): """AutoGain specific error""" @@ -155,145 +170,151 @@ class AutoGain: 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 - init_vel = self.dev.mo1_bragg.velocity + 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) - # TODO, subscribe to the signal, not the property! - # 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 + # Get initial gains for name, ch in active_channels.items(): - lowest_gain = AVAILABLE_GAINS[0] - logger.info(f"Setting gain to {lowest_gain:.0e}".replace("+", "")) - ch["self.dev"].set_gain(f"{lowest_gain:.0e}".replace("+", "")) - ch["gain"] = lowest_gain + ch["init_gain"] = ch["self.dev"].amp.cGain_ENUM.get() - 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} + logger.info("All checks done, start preparing for measurement") - # 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 + try: + logger.info(f"Move mono to start of {emin} eV") + self.dev.mo1_bragg.velocity.put(MONO_VELOCITY) + wait_for_signal(self.dev.mo1_bragg.velocity, MONO_VELOCITY, timeout=TIMEOUT_PUT_PV) 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]["self.dev"].set_gain( - f"{next_gain:.0e}".replace("+", "") - ) - 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]["self.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) + # Set NIDAQ to max mode + self.dev.nidaq.epics_mode.put(EpicsMode.MAX) + wait_for_signal(self.dev.nidaq.epics_mode, EpicsMode.MAX, 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: + # Rest max values of NIDAQ signals + self.dev.nidaq.epics_max_reset.put(True) + time.sleep(0.2) + + # 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 < MIN_RING_CURRENT: + raise AutoGainError( + f"Ring current dropped below {MIN_RING_CURRENT} 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()) + + # 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 < MIN_RING_CURRENT: + raise AutoGainError( + f"Ring current dropped below {MIN_RING_CURRENT} 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]["self.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]["self.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) + except Exception as e: + logger.error("Error during measurement, reset gains to their initial values") + for name, ch in active_channels.items(): + ch["self.dev"].amp.cGain_ENUM.put(ch["init_gain"]) + raise AutoGainError(f"During execution of auto-gain, an exception occured: {e}") from e + # 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_signal(self.dev.nidaq.epics_mode, EpicsMode.MEAN, 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) # TODO: Fix, not using property! - # status = CompareStatus(self.dev.mo1_bragg.velocity, init_vel) - # status.wait(timeout=TIMEOUT_PUT_PV) + self.dev.mo1_bragg.velocity.put(init_vel) + wait_for_signal(self.dev.mo1_bragg.velocity, init_vel, timeout=TIMEOUT_PUT_PV) def _get_ring_current(self) -> float: ring_current = 0 diff --git a/debye_bec/bec_widgets/widgets/scheduler/scheduler.py b/debye_bec/bec_widgets/widgets/scheduler/scheduler.py index 00c5b8b..2bb5b06 100644 --- a/debye_bec/bec_widgets/widgets/scheduler/scheduler.py +++ b/debye_bec/bec_widgets/widgets/scheduler/scheduler.py @@ -488,9 +488,7 @@ class Scheduler(BECWidget, QWidget): ) elif re.match(r"^dev\..*\.move", item.command): if item.form_state["relative"]: - text = ( - f"Move {item.form_state['device_name']} by {item.form_state['value']}" - ) + text = f"Move {item.form_state['device_name']} by {item.form_state['value']:+g}" else: text = ( f"Move {item.form_state['device_name']} to {item.form_state['value']}" @@ -502,7 +500,17 @@ class Scheduler(BECWidget, QWidget): + f"{item.form_state['gas2']} to {item.form_state['pressure']} bar" ) elif item.command.startswith("dev.reffoilchanger"): - text = f"Insert {item.form_state['ref']} reference foil" + text = f"Insert {item.form_state['foil']} reference foil" + elif item.command.startswith("digital_twin.move_with_config({"): + config = item.form_state["config"] + if config["mo1_mode"] == "Monochromatic": + text = f"Align Beamline for {config['energy']} eV" + else: + text = "Align Beamline for pinkbeam" + text += f" (CM: {config['cm_stripe']} at {config['cm_pitch']*1e3} mrad" + if config["mo1_mode"] == "Monochromatic": + text += f", MO1: {config['mo1_xtal']}" + text += f", FM: {config['fm_stripe']} at {config['fm_rotx']*1e3} mrad)" if text == "": text = f"{item.command}" if item.error: diff --git a/debye_bec/devices/ionization_chambers/ionization_chamber.py b/debye_bec/devices/ionization_chambers/ionization_chamber.py index 92a560d..6b8a107 100644 --- a/debye_bec/devices/ionization_chambers/ionization_chamber.py +++ b/debye_bec/devices/ionization_chambers/ionization_chamber.py @@ -14,6 +14,7 @@ from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase from typeguard import typechecked from debye_bec.devices.ionization_chambers.ionization_chamber_enums import ( + GAIN_MAP, AmplifierEnable, AmplifierFilter, AmplifierGain, @@ -120,13 +121,17 @@ class IonizationChamber0(PSIDeviceBase): self.timeout_for_pvwait = 2.5 super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs) - def set_gain(self, gain: float | str) -> None: + def set_gain(self, gain: float) -> None: """Configure the gain setting of the specified channel Args: - gain (Literal['1e6', '1e7', '5e7', '1e8', '1e9']) : Desired gain + gain(float): Amplifier gain to set. Accepts either a float/int or a string, + e.g. 1e6 or "1e6". Must be one of: 1e6, 1e7, 5e7, 1e8, 1e9. + + Raises: + ValueError: If `gain` cannot be converted to a float, or does not + match one of the supported gain values. """ - # TODO: Fix docstring if self.amp.cOnOff.get() == AmplifierEnable.OFF: status = CompareStatus(self.amp.cOnOff, AmplifierEnable.ON) @@ -134,17 +139,21 @@ class IonizationChamber0(PSIDeviceBase): self.amp.cOnOff.put(AmplifierEnable.ON) status.wait(self.timeout_for_pvwait) - match gain: - case "1e6" | 1e6: - self.amp.cGain_ENUM.put(AmplifierGain.G1E6) - case "1e7" | 1e7: - self.amp.cGain_ENUM.put(AmplifierGain.G1E7) - case "5e7" | 5e7: - self.amp.cGain_ENUM.put(AmplifierGain.G5E7) - case "1e8" | 1e8: - self.amp.cGain_ENUM.put(AmplifierGain.G1E8) - case "1e9" | 1e9: - self.amp.cGain_ENUM.put(AmplifierGain.G1E9) + try: + key = float(gain) + except (TypeError, ValueError) as exc: + raise ValueError(f"Invalid gain: {gain!r}") from exc + + try: + enum_val = GAIN_MAP[key] + except KeyError as exc: + raise ValueError( + f"Unsupported gain: {gain!r}. Must be one of {sorted(GAIN_MAP)}" + ) from exc + + self.amp.cGain_ENUM.put(enum_val) + status = CompareStatus(self.amp.cGain_ENUM, enum_val) + status.wait(self.timeout_for_pvwait) def set_filter( self, value: Literal["1us", "3us", "10us", "30us", "100us", "300us", "1ms", "3ms"] diff --git a/debye_bec/devices/ionization_chambers/ionization_chamber_enums.py b/debye_bec/devices/ionization_chambers/ionization_chamber_enums.py index 2cd7ecb..74bff1d 100644 --- a/debye_bec/devices/ionization_chambers/ionization_chamber_enums.py +++ b/debye_bec/devices/ionization_chambers/ionization_chamber_enums.py @@ -19,6 +19,15 @@ class AmplifierGain(int, enum.Enum): G1E9 = 4 +GAIN_MAP = { + 1e6: AmplifierGain.G1E6, + 1e7: AmplifierGain.G1E7, + 5e7: AmplifierGain.G5E7, + 1e8: AmplifierGain.G1E8, + 1e9: AmplifierGain.G1E9, +} + + class AmplifierFilter(int, enum.Enum): """Enum class for the filter of the channel""" diff --git a/debye_bec/devices/nidaq/nidaq_enums.py b/debye_bec/devices/nidaq/nidaq_enums.py index 9b659ee..19adf39 100644 --- a/debye_bec/devices/nidaq/nidaq_enums.py +++ b/debye_bec/devices/nidaq/nidaq_enums.py @@ -59,8 +59,9 @@ class EncoderFactors(int, enum.Enum): X2 = 5 X4 = 6 + class EpicsMode(int, enum.Enum): """Mode when sending through EPICS""" MEAN = 0 - MAX = 0 + MAX = 1