diff --git a/debye_bec/bec_ipython_client/plugins/auto_gain.py b/debye_bec/bec_ipython_client/plugins/auto_gain.py index 846e785..a5ccd5c 100644 --- a/debye_bec/bec_ipython_client/plugins/auto_gain.py +++ b/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) diff --git a/debye_bec/bec_widgets/widgets/digital_twin/widgets/move_widget.py b/debye_bec/bec_widgets/widgets/digital_twin/widgets/move_widget.py index aa57582..0aa6f24 100644 --- a/debye_bec/bec_widgets/widgets/digital_twin/widgets/move_widget.py +++ b/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() diff --git a/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py b/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py index 522aab1..79d6bc3 100644 --- a/debye_bec/bec_widgets/widgets/scheduler/item_dialog.py +++ b/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 == []: diff --git a/debye_bec/device_configs/x01da_experimental_hutch.yaml b/debye_bec/device_configs/x01da_experimental_hutch.yaml index 219e0d0..cec7109 100644 --- a/debye_bec/device_configs/x01da_experimental_hutch.yaml +++ b/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 diff --git a/debye_bec/device_configs/x01da_optics.yaml b/debye_bec/device_configs/x01da_optics.yaml index ea61ae9..c96564d 100644 --- a/debye_bec/device_configs/x01da_optics.yaml +++ b/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 diff --git a/debye_bec/devices/absorber.py b/debye_bec/devices/absorber.py index fa719db..29c3c9f 100644 --- a/debye_bec/devices/absorber.py +++ b/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: diff --git a/debye_bec/devices/eh_shutter.py b/debye_bec/devices/eh_shutter.py index 52d06e3..27b141e 100644 --- a/debye_bec/devices/eh_shutter.py +++ b/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, diff --git a/debye_bec/devices/op_shutter.py b/debye_bec/devices/op_shutter.py index 99b8ba1..4f963ad 100644 --- a/debye_bec/devices/op_shutter.py +++ b/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: