wip

debye_bec/bec_ipython_client/plugins/auto_gain.py

@@ -1,60 +1,63 @@
-""" Module to automatically set the gains for the selected amplifiers"""
+"""Module to automatically set the gains for the selected amplifiers"""
 
-from bisect import bisect_right
-import time
 import builtins
+import time
+from bisect import bisect_right
 
 import xraydb
 from bec_lib import bec_logger
 
-from ...devices.nidaq.nidaq_enums import NidaqState
 from ...devices.absorber import STATUS as ABS_STATUS
-from ...devices.op_shutter import STATUS as OP_PH_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 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
+MIN_RING_CURRENT = 5  # Minimum ring current to use auto-gain
+NOMINAL_RING_CURRENT = 400  # Nominal ring current of SLS2
 
-MONO_VELOCITY = 20 # Move velocity in deg/s
-TIMEOUT_MONO_PV = 5 # Timeout to set a PV on the mono
-TIMEOUT_MONO_MOVE = 30 # Timeout to finish a movement on the mono
+MONO_VELOCITY = 20  # Move velocity in deg/s
+TIMEOUT_MONO_PV = 5  # Timeout to set a PV on the mono
+TIMEOUT_MONO_MOVE = 30  # Timeout to finish a movement on the mono
 
 AVAILABLE_GAINS = [1e6, 1e7, 5e7, 1e8, 1e9]  # ascending order
-MIN_SIGNAL = 0.05 # Minimum signal to count as valid signal
+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"""
+
+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')
+            raise AutoGainError("Did not get dev")
         if bec is None:
-            raise AutoGainError('Did not get bec')
+            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
+        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.
-        
+
         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'
@@ -73,25 +76,25 @@ class AutoGain():
             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 the ring current drops to 0 mA during the measurement (beamdump)
         """
 
         if amplifier is None:
-            amplifier = ['ic0', 'ic1', 'ic2', 'pips']
+            amplifier = ["ic0", "ic1", "ic2", "pips"]
 
         # Make sure NIDAQ is in standby mode
-        if self.dev.nidaq.state.get() is not NidaqState.STANDBY:
-            raise AutoGainError('NIDAQ was not in Standby mode, cannot proceed.')
+        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')
+            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_shutter.status.get() != OP_PH_STATUS.NOT_CLOSED:
-            raise AutoGainError('OP Photon Shutter is closed, no beam')
-        if self.dev.eh_shutter.status.get() != EH_PH_STATUS.NOT_CLOSED:
-            raise AutoGainError('EH Photon Shutter is closed, no beam')
+            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
@@ -101,56 +104,57 @@ class AutoGain():
         # 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}')
+            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, high_limit = self.dev.mo1_bragg.limits()
+        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'
+                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},
+            "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:
+            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() is not True:
+            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.v.get() < 1000:
+                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:
+                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() is not True:
-                    raise AutoGainError(f'Gas filling of ionization chamber {name} is not OK')
+            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')
+        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')
+        logger.info(f"Move mono to start of {emin} eV")
         status = self.dev.mo1_bragg.move(emin)
         status.wait(TIMEOUT_MONO_MOVE)
 
@@ -160,20 +164,20 @@ class AutoGain():
         # 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
+            ch["self.dev"].set_gain(lowest_gain)
+            ch["gain"] = lowest_gain
 
         remeasure = True
-        logger.info(f'Start measurement from {emin} eV to {emax} eV')
+        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')
+            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')
+                raise AutoGainError("Ring current dropped to 0 mA right before measurement")
 
             # Scan range, recording the peak NIDAQ signal per channel
             status = self.dev.mo1_bragg.velocity.put(MONO_VELOCITY)
@@ -181,16 +185,16 @@ class AutoGain():
             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())
+                data[name] = max(data[name], ch["signal"].get())
 
             # Rest max values of NIDAQ signals
             # TODO implement
 
             # Measure current ring current again
             ring_current_2 = self._get_ring_current()
-            logger.info(f'Ring current right after measurement: {ring_current_2} mA')
+            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')
+                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
@@ -200,35 +204,39 @@ class AutoGain():
             remeasure = False
             for name, ch in active_channels.items():
                 raw_signal = data[name]
-                logger.info(f'Raw signal for self.device {name} is {raw_signal} V')
+                logger.info(f"Raw signal for self.device {name} is {raw_signal} V")
                 if comp_ring_current:
                     raw_signal = raw_signal * NOMINAL_RING_CURRENT / ring_current
-                    logger.info(f'Compensate for ring current, new raw signal is {raw_signal} V')
+                    logger.info(f"Compensate for ring current, new raw signal is {raw_signal} V")
                 if raw_signal < MIN_SIGNAL:
-                    logger.info(f'Raw signal for self.device {name} is below {MIN_SIGNAL}')
+                    logger.info(f"Raw signal for self.device {name} is below {MIN_SIGNAL}")
                     # Choose next gain to be 100x the current gain, or if this gain does not exist,
                     # choose the next smaller one
-                    if ch['gain'] == AVAILABLE_GAINS[-1]:
+                    if ch["gain"] == AVAILABLE_GAINS[-1]:
                         logger.warning(
-                            f"Amplifier of {name} at highest gain {ch['gain']} and still not" +
-                            f" measured signal above {MIN_SIGNAL}"
+                            f"Amplifier of {name} at highest gain {ch['gain']} and still not"
+                            + f" measured signal above {MIN_SIGNAL}"
                         )
                     else:
                         next_gain = AVAILABLE_GAINS[
-                            bisect_right(AVAILABLE_GAINS, ch['gain'] * 100) - 1
+                            bisect_right(AVAILABLE_GAINS, ch["gain"] * 100) - 1
                         ]
-                        ch['self.dev'].set_gain(next_gain)
+                        ch["self.dev"].set_gain(next_gain)
                         logger.info(
-                            f'Setting gain of self.device {name} to {next_gain:.0e} and remeasure'
+                            f"Setting gain of self.device {name} to {next_gain:.0e} and remeasure"
                         )
                         remeasure = True
                 else:
                     gain = max(
-                        (g for g in AVAILABLE_GAINS if raw_signal / ch['gain'] * g <= FULL_SCALE_V * SAFETY_MARGIN),
+                        (
+                            g
+                            for g in AVAILABLE_GAINS
+                            if raw_signal / ch["gain"] * g <= FULL_SCALE_V * SAFETY_MARGIN
+                        ),
                         default=min(AVAILABLE_GAINS),
                     )
-                    ch['self.dev'].set_gain(gain)
-                    logger.info(f'Calculated final gain for {name} of {gain:.0e}')
+                    ch["self.dev"].set_gain(gain)
+                    logger.info(f"Calculated final gain for {name} of {gain:.0e}")
 
             # Wait for mono to return to start position
             status.wait(TIMEOUT_MONO_MOVE)

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

@@ -292,15 +292,13 @@ class MotionWorker(QObject):
         Args:
             surveyed_axes (list): List of dictionaries of devices
         """
+        logger.info(f"Move axis {self.motor} to target {self._target}, move_relative={relative}")
         try:
             if alias:
                 self.motor = alias
             if abs_closed:
                 if self.dev.abs.status.get() == ABS_STATUS.OPEN:
                     status = self.dev.abs.close()
-                    # TODO Set timeout to 0.001 and check if it actually raises
-                    # (it should not start motion).
-                    # Check of behavior of digital twin afterwards.
                     status.wait(timeout=5)
             if surveyed_axes is not None:
                 for surv_ax in surveyed_axes:
@@ -336,7 +334,8 @@ class MotionWorker(QObject):
                             self.finished.emit()
                             break
             self.finished.emit()
-        except:
+        except Exception as e:
+            logger.error(f"Error during movement of {self.motor}: {e}")
             self.error.emit()
             self.finished.emit()
 

debye_bec/bec_widgets/widgets/scheduler/item_dialog.py

@@ -54,6 +54,8 @@ from bec_widgets.widgets.control.device_input.device_combobox.device_combobox im
     DeviceComboBox,
 )
 from bec_widgets.widgets.control.scan_control.scan_control import ScanControl, ScanParameterConfig
+
+# pylint: disable=E0611
 from qtpy.QtWidgets import (
     QCheckBox,
     QComboBox,
@@ -71,6 +73,7 @@ from qtpy.QtWidgets import (
     QVBoxLayout,
     QWidget,
 )
+
 from .qt_widgets import MyButton
 
 logger = bec_logger.logger
@@ -86,7 +89,15 @@ _TAB_OTHER = 3
 class ScheduleItemDialog(QDialog):
     """Add or edit one schedule item, via ScanControl, a move form, Digital Twin, or free text."""
 
-    def __init__(self, scans, dev, parent=None, initial: dict | None = None, client=None, beamline:str|None=None):
+    def __init__(
+        self,
+        scans,
+        dev,
+        parent=None,
+        initial: dict | None = None,
+        client=None,
+        beamline: str | None = None,
+    ):
         super().__init__(parent)
         self.setWindowTitle("Schedule item")
         self.setMinimumSize(520, 480)
@@ -96,11 +107,17 @@ class ScheduleItemDialog(QDialog):
         self._client = client
 
         self.beamline = beamline
-        if self.beamline in ['x01da', 'x10da']:
+        if self.beamline in ["x01da", "x10da"]:
+            logger.info(f"Loading bl-specific modules for beamline {self.beamline}")
             from ....bec_ipython_client.plugins.digital_twin.digital_twin import DigitalTwinCore
             from ..digital_twin.digital_twin import DigitalTwin
-            from ..scan_control_xas.scan_control_xas import ScanControlXAS
             from ..edge_selector import EdgeSelector
+            from ..scan_control_xas.scan_control_xas import ScanControlXAS
+
+            self.DigitalTwinCore = DigitalTwinCore
+            self.DigitalTwin = DigitalTwin
+            self.EdgeSelector = EdgeSelector
+            self.ScanControlXAS = ScanControlXAS
 
         layout = QVBoxLayout(self)
         self.tabs = QTabWidget()
@@ -108,7 +125,7 @@ class ScheduleItemDialog(QDialog):
 
         self._build_scan_tab()
         self._build_move_tab()
-        if self.beamline in ['x01da', 'x10da']:
+        if self.beamline in ["x01da", "x10da"]:
             self._build_digital_twin_tab()
         self._build_custom_tab()
 
@@ -137,8 +154,8 @@ class ScheduleItemDialog(QDialog):
         # client=None resolves to the same process-wide BEC client
         # (bec_dispatcher.client) our own widget uses - no second Redis
         # connection is opened.
-        if self.beamline in ['x01da', 'x10da']:
-            self.scan_control = ScanControlXAS(parent=tab, client=self._client)
+        if self.beamline in ["x01da", "x10da"]:
+            self.scan_control = self.ScanControlXAS(parent=tab, client=self._client)
         else:
             self.scan_control = ScanControl(parent=tab, client=self._client)
         self.scan_control.button_run_scan.hide()
@@ -240,7 +257,7 @@ class ScheduleItemDialog(QDialog):
         # doesn't have to grow to match it.
         scroll = QScrollArea()
         scroll.setWidgetResizable(True)
-        self.digital_twin = DigitalTwin(parent=scroll, client=self._client)
+        self.digital_twin = self.DigitalTwin(parent=scroll, client=self._client)
         # Hide move and abs open buttons
         for mover in self.digital_twin.mover.mover_widgets:
             mover.btn_action.hide()
@@ -303,7 +320,7 @@ class ScheduleItemDialog(QDialog):
         self.custom_edit.setPlaceholderText("scans.xas_simple_scan(12000, 14000, 2, 10)")
         layout.addWidget(self.custom_edit)
 
-        if self.beamline in ['x01da', 'x10da']:
+        if self.beamline in ["x01da", "x10da"]:
             abs_form = self._create_abs_form()
             if abs_form is not None:
                 layout.addWidget(abs_form)
@@ -332,6 +349,7 @@ class ScheduleItemDialog(QDialog):
             form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint)
             self.abs_selector = QComboBox()
             self.abs_selector.addItems(["Open", "Force open", "Close"])
+            form.addRow("Action", self.abs_selector)
 
             button_layout = QHBoxLayout()
             generate_cmd = MyButton("Generate command", "default")
@@ -346,12 +364,12 @@ class ScheduleItemDialog(QDialog):
 
     def _generate_abs_command(self):
         match self.abs_selector.currentText():
-            case 'Open':
-                suffix = 'open()'
-            case 'Force open':
-                suffix = 'open(force=True)'
-            case 'Close':
-                suffix = 'close()'
+            case "Open":
+                suffix = "open()"
+            case "Force open":
+                suffix = "open(force=True)"
+            case "Close":
+                suffix = "close()"
         cmd = f"dev.abs.{suffix}"
         self.custom_edit.setText(cmd)
 
@@ -411,6 +429,8 @@ class ScheduleItemDialog(QDialog):
             self.conc1.setValue(100 - new_val)
 
     def _generate_ic_command(self):
+        if self.conc1.value() + self.conc2.value() != 100:
+            return
         match self.ic_selector.currentText():
             case "IC0":
                 ic = "ic0"
@@ -465,22 +485,26 @@ class ScheduleItemDialog(QDialog):
         edge_selector_layout.addWidget(edge_selector_label)
         edge_selector_layout.addWidget(self.edge_selector_button)
         edge_selector_layout.addWidget(self.edge_label)
+        edge_selector_layout.addStretch()
         self.edge_element = None
         self.edge_edge = None
 
         layout.addLayout(edge_selector_layout)
 
+        layout.addWidget(QLabel("Auto gain on"))
+
         form = QFormLayout()
+        layout.addLayout(form)
         form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.FieldsStayAtSizeHint)
 
         self.ic0_check = QCheckBox("")
-        form.addRow("Auto gain on IC0", self.ic0_check)
+        form.addRow("IC0", self.ic0_check)
         self.ic1_check = QCheckBox("")
-        form.addRow("          on IC1", self.ic1_check)
+        form.addRow("IC1", self.ic1_check)
         self.ic2_check = QCheckBox("")
-        form.addRow("          on IC2", self.ic2_check)
+        form.addRow("IC2", self.ic2_check)
         self.pips_check = QCheckBox("")
-        form.addRow("          on PIPS", self.pips_check)
+        form.addRow("PIPS", self.pips_check)
 
         button_layout = QHBoxLayout()
         generate_cmd = MyButton("Generate command", "default")
@@ -496,11 +520,11 @@ class ScheduleItemDialog(QDialog):
     def _update_edge(self, *_):
         match self.beamline:
             case "x01da":
-                dlg = EdgeSelector(self)
+                dlg = self.EdgeSelector(self)
             case "x10da":
-                dlg = EdgeSelector(self)
+                dlg = self.EdgeSelector(self)
             case _:
-                dlg = EdgeSelector(self)
+                dlg = self.EdgeSelector(self)
         if dlg.exec_():
             self.edge_energy = dlg.selected_energy
             self.edge_label.setText(
@@ -513,7 +537,12 @@ class ScheduleItemDialog(QDialog):
         if self.edge_edge is None or self.edge_element is None:
             return
         amplifiers = []
-        for amp, name in [(self.ic0_check, 'ic0'), (self.ic1_check, 'ic1'), (self.ic2_check, 'ic2'), (self.pips_check, 'pips')]:
+        for amp, name in [
+            (self.ic0_check, "ic0"),
+            (self.ic1_check, "ic1"),
+            (self.ic2_check, "ic2"),
+            (self.pips_check, "pips"),
+        ]:
             if amp.isChecked():
                 amplifiers.append(name)
         if amplifiers == []:

debye_bec/device_configs/x01da_experimental_hutch.yaml

@@ -3,7 +3,7 @@
 ## Experimental Hutch Photon Shutter ##
 #######################################
 
-eh-sh:
+eh_sh:
   readoutPriority: baseline
   description: Experimental Hutch Photon Shutter
   deviceClass: debye_bec.devices.eh_shutter.EHPhotonShutter

debye_bec/device_configs/x01da_optics.yaml

@@ -3,7 +3,7 @@
 ##     Optics Photon Shutter     ##
 ###################################
 
-op-sh:
+op_sh:
   readoutPriority: baseline
   description: Optics Hutch Photon Shutter
   deviceClass: debye_bec.devices.op_shutter.OPPhotonShutter

debye_bec/devices/absorber.py

@@ -17,6 +17,7 @@ if TYPE_CHECKING:
 
 TIMEOUT_FOR_PV = 5
 
+
 class AbsorberError(Exception):
     """Absorber specific exception"""
 
@@ -40,6 +41,7 @@ class STATUS(int, enum.Enum):
     MAN_OPEN = 13
     UNDEFINED = 14
 
+
 class BL_ENABLE(int, enum.Enum):
     """Beamline enable"""
 
@@ -64,7 +66,7 @@ class Absorber(PSIDeviceBase):
         string=True,
         doc="Absorber Status",
     )
-    close4bl = Cpt(EpicsSignal, suffix='CLOSE4BL', kind='config', doc='Beamline enable')
+    close4bl = Cpt(EpicsSignal, suffix="CLOSE4BL", kind="config", doc="Beamline enable")
 
     def __init__(self, *, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs):
         super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
@@ -73,15 +75,15 @@ class Absorber(PSIDeviceBase):
         # Wait for connection on all components, ensure IOC is connected
         self.wait_for_connection(all_signals=True, timeout=5)
 
-    def open(self, force:bool=False) -> DeviceStatus | None:
+    def open(self, force: bool = False) -> DeviceStatus | None:
         """Open the Absorber
 
         Args:
         force(bool): If needed, set bl status to enable and bl enable to ENABLE, defaults to False
-        
+
         """
-        if force and self.device_manager.devices.get('bl_status', None) is None:
-            raise AbsorberError('bl_status is not in device config, thus cannot use force = True')
+        if force and self.device_manager.devices.get("bl_status", None) is None:
+            raise AbsorberError("bl_status is not in device config, thus cannot use force = True")
         if self.status.get() == STATUS.CLOSED:
             if force:
                 if self.device_manager.bl_status.get() == BlStatus.OFFLINE:

debye_bec/devices/eh_shutter.py

@@ -17,12 +17,14 @@ if TYPE_CHECKING:
 
 TIMEOUT_FOR_PV = 5
 
+
 class STATUS(int, enum.Enum):
     """Shutter States"""
 
     NOT_CLOSED = 0
     CLOSED = 1
 
+
 class BL_ENABLE(int, enum.Enum):
     """Beamline enable"""
 
@@ -30,15 +32,23 @@ class BL_ENABLE(int, enum.Enum):
     ENABLE = 1
 
 
-class OPPhotonShutter(PSIDeviceBase):
+class EHPhotonShutter(PSIDeviceBase):
     """Class for the Experimental Hutch Photon Shutter"""
 
     USER_ACCESS = ["open", "close"]
 
-    request_open = Cpt(EpicsSignal, suffix="EH1-PSYS:SH-A-OPEN-SET", kind="config", doc="Open Shutter")
-    request_close = Cpt(EpicsSignal, suffix="EH1-PSYS:SH-A-CLOSE-SET", kind="config", doc="Close Shutter")
+    request_open = Cpt(
+        EpicsSignal, suffix="EH1-PSYS:SH-A-OPEN-SET", kind="config", doc="Open Shutter"
+    )
+    request_close = Cpt(
+        EpicsSignal, suffix="EH1-PSYS:SH-A-CLOSE-SET", kind="config", doc="Close Shutter"
+    )
     status = Cpt(
-        EpicsSignalRO, suffix="EH1", kind="normal", auto_monitor=True, doc="Shutter Status"
+        EpicsSignalRO,
+        suffix="EH1-PSYS:SH-A-CLOSE",
+        kind="normal",
+        auto_monitor=True,
+        doc="Shutter Status",
     )
     status_string = Cpt(
         EpicsSignalRO,

debye_bec/devices/op_shutter.py

@@ -17,6 +17,7 @@ if TYPE_CHECKING:
 
 TIMEOUT_FOR_PV = 5
 
+
 class OPPhotonShutterError(Exception):
     """Shutter specific exception"""
 
@@ -27,6 +28,7 @@ class STATUS(int, enum.Enum):
     NOT_CLOSED = 0
     CLOSED = 1
 
+
 class BL_ENABLE(int, enum.Enum):
     """Beamline enable"""
 
@@ -39,10 +41,18 @@ class OPPhotonShutter(PSIDeviceBase):
 
     USER_ACCESS = ["open", "close"]
 
-    request_open = Cpt(EpicsSignal, suffix="OP-PSYS:SH-A-OPEN-SET", kind="config", doc="Open Shutter")
-    request_close = Cpt(EpicsSignal, suffix="OP-PSYS:SH-A-CLOSE-SET", kind="config", doc="Close Shutter")
+    request_open = Cpt(
+        EpicsSignal, suffix="OP-PSYS:SH-A-OPEN-SET", kind="config", doc="Open Shutter"
+    )
+    request_close = Cpt(
+        EpicsSignal, suffix="OP-PSYS:SH-A-CLOSE-SET", kind="config", doc="Close Shutter"
+    )
     status = Cpt(
-        EpicsSignalRO, suffix="OP-PSYS:SH-A-CLOSE", kind="normal", auto_monitor=True, doc="Shutter Status"
+        EpicsSignalRO,
+        suffix="OP-PSYS:SH-A-CLOSE",
+        kind="normal",
+        auto_monitor=True,
+        doc="Shutter Status",
     )
     status_string = Cpt(
         EpicsSignalRO,
@@ -52,7 +62,7 @@ class OPPhotonShutter(PSIDeviceBase):
         string=True,
         doc="Shutter Status",
     )
-    close4bl = Cpt(EpicsSignal, suffix='FE-BST1:CLOSE4BL', kind='config', doc='Beamline enable')
+    close4bl = Cpt(EpicsSignal, suffix="FE-BST1:CLOSE4BL", kind="config", doc="Beamline enable")
 
     def __init__(self, *, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs):
         super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
@@ -61,15 +71,17 @@ class OPPhotonShutter(PSIDeviceBase):
         # Wait for connection on all components, ensure IOC is connected
         self.wait_for_connection(all_signals=True, timeout=5)
 
-    def open(self, force:bool=False) -> DeviceStatus | None:
+    def open(self, force: bool = False) -> DeviceStatus | None:
         """Open the Shutter
 
         Args:
         force(bool): If needed, set bl status to enable and bl enable to ENABLE, defaults to False
-        
+
         """
-        if force and self.device_manager.devices.get('bl_status', None) is None:
-            raise OPPhotonShutterError('bl_status is not in device config, thus cannot use force = True')
+        if force and self.device_manager.devices.get("bl_status", None) is None:
+            raise OPPhotonShutterError(
+                "bl_status is not in device config, thus cannot use force = True"
+            )
         if self.status.get() == STATUS.CLOSED:
             if force:
                 if self.device_manager.bl_status.get() == BlStatus.OFFLINE: