feat(auto_gain): Implement for X10DA

debye_bec/bec_ipython_client/plugins/auto_gain_core.py

@@ -12,21 +12,23 @@ 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
+from .beamline import BeamlineId, get_beamline_id
 
 logger = bec_logger.logger
 
-EMIN = -100
-EMAX = 200
+EMIN = -200  # Minimum energy will be edge energy - EMIN
+EMAX = 1000  # Maximum energy will be edge energy + EMAX
 
-MIN_RING_CURRENT = -1  # 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
 
-MONO_VELOCITY = 20  # Move velocity in deg/s
+MONO_VELOCITY = 20  # Move velocity in deg/s (only X01DA)
 TIMEOUT_MONO_MOVE = 30  # Timeout to finish a movement on the mono
 
-AVAILABLE_GAINS = [1e6, 1e7, 5e7, 1e8, 1e9]  # ascending order
+AVAILABLE_GAINS_X01DA = [1e6, 1e7, 5e7, 1e8, 1e9]  # ascending order
+AVAILABLE_GAINS_X10DA = [1e6, 1e7, 1e8, 1e9, 1e10]  # 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
@@ -64,17 +66,22 @@ class AutoGainCore:
             raise AutoGainError("Did not get bec")
         self.dev = dev
         self.bec = bec
+        self.beamline = get_beamline_id()
+        if self.beamline == BeamlineId.X01DA:
+            self.available_gains = AVAILABLE_GAINS_X01DA
+        else:
+            self.available_gains = AVAILABLE_GAINS_X10DA
         self.linked_channels = None
         self.prohibited_gains = {}
 
-    def available_gains(self, amplifier):
+    def allowed_gains(self, amplifier):
         """Returns a list of gains which are allowed for the amplifier"""
         if amplifier in self.prohibited_gains:
-            gains = AVAILABLE_GAINS.copy()
+            gains = self.available_gains.copy()
             for prohibited_gain in self.prohibited_gains[amplifier]:
                 gains.remove(prohibited_gain)
             return gains
-        return AVAILABLE_GAINS
+        return self.available_gains
 
     def start(
         self,
@@ -96,7 +103,9 @@ class AutoGainCore:
             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']
+                                  Defaults to all amplifiers ->
+                                  X01DA = ['ic0', 'ic1', 'ic2', 'pips']
+                                  X10DA = ['ic0', 'ic1', 'ic2']
             link_ic0_ic1(bool): Apply same gain to ic0 and ic1. Defaults to True
             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
@@ -127,7 +136,10 @@ class AutoGainCore:
             self.prohibited_gains = prohibited_gains
 
         if amplifier is None:
-            amplifier = ["ic0", "ic1", "ic2", "pips"]
+            if self.beamline == BeamlineId.X01DA:
+                amplifier = ["ic0", "ic1", "ic2", "pips"]
+            else:
+                amplifier = ["ic0", "ic1", "ic2"]
 
         # Make sure NIDAQ is in standby mode
         if self.dev.nidaq.state.get() != NidaqState.STANDBY:
@@ -155,15 +167,17 @@ class AutoGainCore:
         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"
-            )
+        # Check range of mono (only X01DA)
+        if self.beamline == BeamlineId.X01DA:
+            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"
+                )
+        # X10DA energy axis (goniometer) does not have lo/hi limits
 
         # Map amplifier names to their NIDAQ channels
         channel_map = {
@@ -174,30 +188,38 @@ class AutoGainCore:
         }
         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")
+        # Tests only for X01DA:
+        if self.beamline == BeamlineId.X01DA:
+            # Check if amplifieres are switched on (only X01DA)
+            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 high voltage on ionization chambers (only X01DA)
+            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")
+            # Check gas filling of ionization chambers (only X01DA)
+            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")
 
         # Get initial monochromator position and velocity
-        init_pos = self.dev.mo1_bragg.position
-        init_vel = self.dev.mo1_bragg.velocity.get()
+        if self.beamline == BeamlineId.X01DA:
+            init_pos = self.dev.mo1_bragg.position
+            init_vel = self.dev.mo1_bragg.velocity.get()
+        else:
+            init_pos = self.dev.mo1_rotx_energy.get()
+            init_vel = None
 
         # Get initial gains
         for name, ch in active_channels.items():
@@ -209,9 +231,12 @@ class AutoGainCore:
 
         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)
+            if self.beamline == BeamlineId.X01DA:
+                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)
+            else:
+                status = self.dev.mo1_rotx_energy.move(emin)
             status.wait(TIMEOUT_MONO_MOVE)
 
             # Set NIDAQ to max mode
@@ -220,7 +245,7 @@ class AutoGainCore:
 
             # Set gains to lowest gain
             for name, ch in active_channels.items():
-                lowest_gain = self.available_gains(name)[0]
+                lowest_gain = self.allowed_gains(name)[0]
                 ch["self.dev"].set_gain(lowest_gain)
                 ch["gain"] = lowest_gain
 
@@ -245,8 +270,10 @@ class AutoGainCore:
                     )
 
                 # 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)
+                if self.beamline == BeamlineId.X01DA:
+                    self.dev.mo1_bragg.move(emax).wait(timeout=TIMEOUT_MONO_MOVE)
+                else:
+                    self.dev.mo1_rotx_energy.move(emax).wait(timeout=TIMEOUT_MONO_MOVE)
                 for name, ch in active_channels.items():
                     data[name] = max(data[name], ch["signal"].get())
 
@@ -260,7 +287,10 @@ class AutoGainCore:
                 ring_current = (ring_current_1 + ring_current_2) / 2
 
                 # Move back to first monochromator position
-                status = self.dev.mo1_bragg.move(emin)
+                if self.beamline == BeamlineId.X01DA:
+                    status = self.dev.mo1_bragg.move(emin)
+                else:
+                    status = self.dev.mo1_rotx_energy.move(emin)
 
                 # Choose gain per channel based on the max signal recorded during the scan
                 remeasure = False
@@ -297,14 +327,14 @@ class AutoGainCore:
 
                     if raw_signal_min < MIN_SIGNAL:
                         logger.info(f"Raw signal for {label} is below {MIN_SIGNAL}")
-                        if gain_ref == self.available_gains(name)[-1]:
+                        if gain_ref == self.allowed_gains(name)[-1]:
                             logger.warning(
                                 f"Amplifier of {label} at highest gain {gain_ref} and still not"
                                 + f" measured signal above {MIN_SIGNAL}"
                             )
                         else:
-                            next_gain = self.available_gains(name)[
-                                bisect_right(self.available_gains(name), gain_ref * 100) - 1
+                            next_gain = self.allowed_gains(name)[
+                                bisect_right(self.allowed_gains(name), gain_ref * 100) - 1
                             ]
                             for gname in group_names:
                                 active_channels[gname]["self.dev"].set_gain(next_gain)
@@ -316,10 +346,10 @@ class AutoGainCore:
                         gain = max(
                             (
                                 g
-                                for g in self.available_gains(name)
+                                for g in self.allowed_gains(name)
                                 if raw_signal_max / gain_ref * g <= FULL_SCALE_V * SAFETY_MARGIN
                             ),
-                            default=min(self.available_gains(name)),
+                            default=min(self.allowed_gains(name)),
                         )
                         for gname in group_names:
                             active_channels[gname]["self.dev"].set_gain(gain)
@@ -345,10 +375,14 @@ class AutoGainCore:
         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)
-        wait_for_signal(self.dev.mo1_bragg.velocity, init_vel, timeout=TIMEOUT_PUT_PV)
+        if self.beamline == BeamlineId.X01DA:
+            status = self.dev.mo1_bragg.move(init_pos)
+            status.wait(TIMEOUT_MONO_MOVE)
+            self.dev.mo1_bragg.velocity.put(init_vel)
+            wait_for_signal(self.dev.mo1_bragg.velocity, init_vel, timeout=TIMEOUT_PUT_PV)
+        else:
+            status = self.dev.mo1_rotx_energy.move(init_pos)
+            status.wait(TIMEOUT_MONO_MOVE)
 
         return results
 

debye_bec/bec_ipython_client/plugins/beamline.py

@@ -1,12 +1,20 @@
 import socket
+from enum import Enum
 
 from bec_lib import bec_logger
 
-from .types import BeamlineId
-
 logger = bec_logger.logger
 
 
+class BeamlineId(str, Enum):
+    """
+    Identifier for supported beamlines.
+    """
+
+    X01DA = "x01da"
+    X10DA = "x10da"
+
+
 def get_beamline_id() -> BeamlineId:
     """
     Based on the bec servers hostname, tries to extract the beamline
@@ -38,14 +46,3 @@ def get_beamline_id() -> BeamlineId:
                 raise ValueError(f"Wrong selection {bl}")
         else:
             raise ValueError("Cannot open digital twin without a beamline")
-
-
-def get_parameters():
-    beamline = get_beamline_id()
-    if beamline == "x01da":
-        from . import x01da_parameters as parameters
-    elif beamline == "x10da":
-        from . import x10da_parameters as parameters
-    else:
-        raise ValueError(f"Unknown beamline: {beamline}")
-    return parameters

debye_bec/bec_ipython_client/plugins/digital_twin_core/__init__.py

@@ -1,3 +1,3 @@
-from .beamline import get_parameters
+from .digital_twin_core import get_parameters
 
 parameters = get_parameters()

debye_bec/bec_ipython_client/plugins/digital_twin_core/digital_twin_core.py

@@ -10,8 +10,8 @@ from bec_lib import bec_logger
 from scipy.interpolate import UnivariateSpline
 from xrt.backends.raycing.physconsts import AVOGADRO, CHeVcm
 
+from ..beamline import get_beamline_id
 from . import parameters as bl
-from .beamline import get_beamline_id
 from .motion_worker import MotionOutcome, MotionWorker
 from .types import BeamlineId, ConfigDict, DataDict, SurfaceDict
 
@@ -26,6 +26,17 @@ OFFSET_FILE_X10DA = Path(__file__).with_name("x10da_offsets.yaml")
 logger = bec_logger.logger
 
 
+def get_parameters():
+    beamline = get_beamline_id()
+    if beamline == "x01da":
+        from . import x01da_parameters as parameters
+    elif beamline == "x10da":
+        from . import x10da_parameters as parameters
+    else:
+        raise ValueError(f"Unknown beamline: {beamline}")
+    return parameters
+
+
 class _ExclusiveGroupManager:
     def __init__(self):
         self._cond = threading.Condition()

debye_bec/bec_ipython_client/plugins/digital_twin_core/motion_worker.py

@@ -8,7 +8,7 @@ from bec_lib import bec_logger
 
 # pylint: disable=E0402
 from ....devices.absorber import STATUS as ABS_STATUS
-from .types import BeamlineId
+from ..beamline import BeamlineId
 
 logger = bec_logger.logger
 

debye_bec/bec_widgets/widgets/auto_gain/auto_gain.py

@@ -5,17 +5,17 @@ Auto Gain: Custom BEC widget to automatically set gains.
 import sys
 
 from bec_lib import bec_logger
-from bec_lib.logger import bec_logger
 from bec_widgets.utils.bec_dispatcher import BECDispatcher
 from bec_widgets.utils.bec_widget import BECWidget
 from bec_widgets.utils.colors import apply_theme, get_accent_colors
 from bec_widgets.utils.error_popups import SafeSlot
-from qtpy.QtCore import QObject, QThread, Signal
 
 # pylint: disable=E0611
+from qtpy.QtCore import QObject, QThread, Signal
 from qtpy.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget
 
 from ....bec_ipython_client.plugins.auto_gain_core import AutoGainCore
+from ....bec_ipython_client.plugins.beamline import BeamlineId, get_beamline_id
 from ..edge_selector import EdgeSelector
 from .qt_widgets import CheckBox, Group, MatrixSelector, MyButton, TextIndicator
 
@@ -67,6 +67,8 @@ class AutoGain(BECWidget, QWidget):
         super().__init__(parent=parent, *arg, **kwargs)
         self.get_bec_shortcuts()
 
+        self.beamline = get_beamline_id()
+
         central = QWidget()
         self.root_layout = QVBoxLayout(central)
 
@@ -85,16 +87,22 @@ class AutoGain(BECWidget, QWidget):
         self.ic1_check = CheckBox("ic1", "IC1")
         self.ic2_check = CheckBox("ic2", "IC2")
         self.pips_check = CheckBox("pips", "PIPS")
-        self.apply_group = Group(
-            "Apply auto gain on", [self.ic0_check, self.ic1_check, self.ic2_check, self.pips_check]
-        )
+        if self.beamline == BeamlineId.X01DA:
+            self.apply_group = Group(
+                "Apply auto gain on",
+                [self.ic0_check, self.ic1_check, self.ic2_check, self.pips_check],
+            )
+        else:  # PIPS of X10DA has fixed gain amplifier
+            self.apply_group = Group(
+                "Apply auto gain on", [self.ic0_check, self.ic1_check, self.ic2_check]
+            )
         self.root_layout.addWidget(self.apply_group)
 
         self.link_ic0_ic1 = CheckBox("link_ic0_ic1", "Link IC0/IC1 gains")
 
         self.comp_ring_current = CheckBox("comp_ring_current", "Calculate for 400 mA ring current")
 
-        self.prohibited_gains = MatrixSelector("Prohibited gains")
+        self.prohibited_gains = MatrixSelector("Prohibited gains", beamline=self.beamline)
 
         self.expert_group = Group(
             "Expert settings", [self.link_ic0_ic1, self.comp_ring_current, self.prohibited_gains]

debye_bec/bec_widgets/widgets/auto_gain/qt_widgets.py

@@ -14,6 +14,8 @@ from qtpy.QtWidgets import (
     QWidget,
 )
 
+from ....bec_ipython_client.plugins.beamline import BeamlineId
+
 LABEL_WIDTH = 118
 ROW_MARGINS = (4, 0, 4, 0)
 ROW_SPACING = 6
@@ -83,11 +85,21 @@ class MatrixSelector(QWidget):
         {'ic0': [1000000.0], 'ic1': [1000000.0, 50000000.0]}
     """
 
-    ROWS = ["IC0", "IC1", "IC2", "PIPS"]
-    COLS = ["1e6", "1e7", "5e7", "1e8", "1e9"]
+    ROWS_X01DA = ["IC0", "IC1", "IC2", "PIPS"]
+    COLS_X01DA = ["1e6", "1e7", "5e7", "1e8", "1e9"]
 
-    def __init__(self, label: str = "", parent=None):
+    ROWS_X10DA = ["IC0", "IC1", "IC2"]
+    COLS_X10DA = ["1e6", "1e7", "1e8", "1e9", "1e10"]
+
+    def __init__(self, label: str, beamline: BeamlineId, parent=None):
         super().__init__(parent)
+        if beamline == BeamlineId.X01DA:
+            self.cols = self.COLS_X01DA
+            self.rows = self.ROWS_X01DA
+        else:
+            self.cols = self.COLS_X10DA
+            self.rows = self.ROWS_X10DA
+
         layout = QHBoxLayout(self)
         layout.setContentsMargins(*ROW_MARGINS)
         layout.setSpacing(ROW_SPACING)
@@ -99,14 +111,14 @@ class MatrixSelector(QWidget):
         grid_layout = QGridLayout()
 
         # Column headers
-        for c, col in enumerate(self.COLS, start=1):
+        for c, col in enumerate(self.cols, start=1):
             grid_layout.addWidget(QLabel(col), 0, c, alignment=Qt.AlignmentFlag.AlignCenter)
 
         # Row headers + checkboxes
         self.checks = {}
-        for r, row in enumerate(self.ROWS, start=1):
+        for r, row in enumerate(self.rows, start=1):
             grid_layout.addWidget(QLabel(row), r, 0)
-            for c, col in enumerate(self.COLS, start=1):
+            for c, col in enumerate(self.cols, start=1):
                 cb = QCheckBox()
                 grid_layout.addWidget(cb, r, c, alignment=Qt.AlignmentFlag.AlignCenter)
                 self.checks[(row, col)] = cb
@@ -117,8 +129,8 @@ class MatrixSelector(QWidget):
     def selection(self):
         """Return e.g. {"ic0": [1e6], "ic1": [1e6, 5e7]}; rows with no selection are omitted."""
         result = {}
-        for row in self.ROWS:
-            values = [float(col) for col in self.COLS if self.checks[(row, col)].isChecked()]
+        for row in self.rows:
+            values = [float(col) for col in self.cols if self.checks[(row, col)].isChecked()]
             if values:
                 result[row.lower()] = values
         return result
@@ -131,9 +143,9 @@ class MatrixSelector(QWidget):
         """
         # Normalize keys to lowercase and values to floats for comparison
         wanted = {key.lower(): {float(v) for v in values} for key, values in selection.items()}
-        for row in self.ROWS:
+        for row in self.rows:
             row_values = wanted.get(row.lower(), set())
-            for col in self.COLS:
+            for col in self.cols:
                 cb = self.checks[(row, col)]
                 cb.setChecked(float(col) in row_values)
 

debye_bec/bec_widgets/widgets/digital_twin/panels/input_panel.py

@@ -7,7 +7,7 @@ from typing import Union
 # pylint: disable=E0611
 from qtpy.QtWidgets import QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget
 
-from .....bec_ipython_client.plugins.digital_twin_core.types import BeamlineId
+from .....bec_ipython_client.plugins.beamline import BeamlineId
 from ..widgets.qt_widgets import (
     Button,
     ComboBox,

debye_bec/bec_widgets/widgets/digital_twin/panels/mover_panel.py

@@ -7,7 +7,7 @@ from typing import Literal
 # pylint: disable=E0611
 from qtpy.QtWidgets import QVBoxLayout, QWidget
 
-from .....bec_ipython_client.plugins.digital_twin_core.types import BeamlineId
+from .....bec_ipython_client.plugins.beamline import BeamlineId
 from ..widgets.move_widget import AbsorberWidget, MoveWidget
 from ..widgets.qt_widgets import Button, Group, QLabel
 

debye_bec/bec_widgets/widgets/digital_twin/widgets/move_widget.py

@@ -18,7 +18,6 @@ from .....bec_ipython_client.plugins.digital_twin_core.motion_worker import (
     MotionOutcome,
     MotionWorker,
 )
-from .....bec_ipython_client.plugins.digital_twin_core.types import BeamlineId
 
 logger = bec_logger.logger