wip

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

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:

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"]

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"""
 

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