wip

debye_bec/bec_ipython_client/plugins/auto_gain.py

@@ -0,0 +1,211 @@
+
+import xraydb
+from bisect import bisect_right
+import time
+import builtins
+import numpy as np
+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
+
+if builtins.__dict__.get("bec") is not None:
+    bec = builtins.__dict__.get("bec")
+    dev = builtins.__dict__.get("dev")
+    scans = builtins.__dict__.get("scans")
+
+logger = bec_logger.logger
+
+EMIN = -100
+EMAX = 200
+
+MIN_RING_CURRENT = 5
+NOMINAL_RING_CURRENT = 400
+
+MONO_VELOCITY = 20
+TIMEOUT_MONO_PV = 5
+TIMEOUT_MONO_MOVE = 60
+
+AVAILABLE_GAINS = [1e6, 1e7, 5e7, 1e8, 1e9]  # 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
+
+class AutoGainError(Exception):
+    """AutoGain specific error"""
+
+class AutoGain():
+    def __init__(self):
+        pass
+
+    def start(self, element:str, edge:str, amplifier:list[str] | None=None, comp_ring_current:bool=True):
+
+        if amplifier is None:
+            amplifier = ['ic0', 'ic1', 'ic2', 'pips']
+
+        # Make sure NIDAQ is in standby mode
+        if dev.nidaq.state.get() is not 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')
+        if dev.abs.status.get() != ABS_STATUS.OPEN:
+            raise AutoGainError('Absorber is closed, no beam')
+        if dev.op_shutter.status.get() != OP_PH_STATUS.NOT_CLOSED:
+            raise AutoGainError('OP Photon Shutter is closed, no beam')
+        if dev.eh_shutter.status.get() != EH_PH_STATUS.NOT_CLOSED:
+            raise AutoGainError('EH Photon Shutter is closed, no beam')
+
+        # Check if no scan is running
+        scan_id = bec.queue.scan_storage.current_scan_id
+        if len(scan_id) > 0:
+            raise AutoGainError(f"Scan with ID {scan_id} is currently running, cannot continue")
+
+        # 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}')
+        emin = energy + EMIN
+        emax = energy + EMAX
+
+        # Check range of mono
+        low_limit, high_limit = dev.mo1_bragg.limits()
+        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'
+            )
+
+        # Map amplifier names to their NIDAQ channels
+        channel_map = {
+            'ic0': {'signal': dev.nidaq.ai0, 'dev': dev.ic0},
+            'ic1': {'signal': dev.nidaq.ai2, 'dev': dev.ic1},
+            'ic2': {'signal': dev.nidaq.ai4, 'dev': dev.ic2},
+            'pips': {'signal': dev.nidaq.ai6, 'dev': 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['dev'].amp.cOnOff.get() != AmplifierEnable.ON:
+                raise AutoGainError(f"Amplifier of device {name} is not enabled")
+
+        # Check high voltage on ionization chambers
+        for name, ch in active_channels.items():
+            if name != 'pips':
+                if ch['dev'].hv_en.ena.get() is not True:
+                    raise AutoGainError(f"High voltage of ionization chamber {name} is not enabled")
+                if ch['dev'].hv.v.get() < 1000:
+                    raise AutoGainError(f"HV voltage of ionization chamber {name} is < 1000")
+                if ch['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['dev'].gmes.status.get() != True:
+                    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 = dev.mo1_bragg.position.get()
+        init_vel = dev.mo1_bragg.velocity.get()
+
+        logger.info(f'Move mono to start of {emin} eV')
+        status = dev.mo1_bragg.move(emin)
+        status.wait(TIMEOUT_MONO_MOVE)
+
+        # Set NIDAQ to max mode
+        # TODO implement
+
+        # Set gains to lowest gain
+        for name, ch in active_channels.items():
+            lowest_gain = AVAILABLE_GAINS[0]
+            ch['dev'].set_gain(lowest_gain)
+            ch['gain'] = lowest_gain
+
+        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}
+
+            # 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
+            status = dev.mo1_bragg.velocity.put(MONO_VELOCITY)
+            status.wait(TIMEOUT_MONO_PV)
+            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
+            # 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')
+            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
+            status = dev.mo1_bragg.move(emin)
+
+            # Choose gain per channel based on the max signal recorded during the scan
+            remeasure = False
+            for name, ch in active_channels.items():
+                raw_signal = data[name]
+                logger.info(f'Raw signal for 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')
+                if raw_signal < MIN_SIGNAL:
+                    logger.info(f'Raw signal for 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]:
+                        logger.warning(f"Amplifier of {name} at highest gain {ch['gain']} and still not measured signal above {MIN_SIGNAL}")
+                    else:
+                        next_gain = AVAILABLE_GAINS[bisect_right(AVAILABLE_GAINS, ch['gain'] * 100) - 1]
+                        ch['dev'].set_gain(next_gain)
+                        logger.info(f'Setting gain of 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),
+                        default=min(AVAILABLE_GAINS),
+                    )
+                    ch['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)
+
+        # Reset NIDAQ to mean mode
+        # TODO implement
+
+        # Wait for mono to move to initial position and reset velocity
+        status = dev.mo1_bragg.move(init_pos)
+        status.wait(TIMEOUT_MONO_MOVE)
+        dev.mo1_bragg.velocity.put(init_vel)
+
+    @staticmethod
+    def _get_ring_current() -> float:
+        ring_current = 0
+        retries = 0
+        while ring_current == 0 and retries < 10:
+            ring_current = dev.curr.get()
+            retries += 1
+            time.sleep(0.01)
+        return ring_current

debye_bec/bec_ipython_client/plugins/move_to_label.py

@@ -1,82 +0,0 @@
-from __future__ import annotations
-
-import builtins
-from typing import TYPE_CHECKING
-
-from bec_lib import bec_logger
-from debye_bec.devices.absorber import STATUS as ABS_STATUS
-
-logger = bec_logger.logger
-# import builtins to avoid linter errors
-dev = builtins.__dict__.get("dev")
-
-class MoveToLabelError(Exception):
-    """Exception for the MoveToLabel function"""
-
-def move_to_label():
-    """
-    Function to move several motors to a specific position defined in the label dict.
-    """
-
-    label = get_device_conditions(label="digitalTwin")
-
-    # Get absorber status and close if open
-    logger.info("Check Frontend Absorber Status")
-    abs_was_open = dev.abs.status.get() == ABS_STATUS.OPEN
-    if abs_was_open:
-        logger.info("  Close Frontend Absorber")
-        status = dev.abs.close()
-        status.wait()
-
-    # Move Frontend Slits
-    logger.info("Move Frontend Slits into position")
-    devices = ["sldi_centerx", "sldi_centery", "sldi_gapx", "sldi_gapy"]
-    matches = {key: label[key] for key in devices if key in label}
-    statuses = []
-    for device in matches.values():
-        statuses.append(device['device'].move(device['value']))
-    for status in statuses:
-        status.wait(timeout=30)
-
-    # Move Collimating mirror
-    logger.info("Move Collimating Mirror into position")
-    if "cm_rotx" in label: # pitch
-        logger.info("  Move pitch into position")
-        surveyed_movement(
-            axis=label['cm_rotx'],
-            surveyed_axes= [
-                {'device': dev.cm_rotz, 'abs_tol': 0.1},
-            ]
-        )
-
-    # Restore absorber position
-    logger.info("Restore Frontend Absorber Status")
-    if abs_was_open:
-        status = dev.abs.open()
-        status.wait()
-
-
-def surveyed_movement(axis, surveyed_axes):
-    """
-    Moves an axis while surverying a set of axes.
-
-    Args:
-        axis (DeviceCondition): Device condition
-        surveyed_axes (list): List of dicts (same format as DeviceCondition)
-
-    Raises:
-        If during movement of axis, one of the surveyed axes moves out of tolerance.
-    """
-
-    for surv_ax in surveyed_axes:
-        surv_ax['old_value'] = surv_ax['device'].read()
-    status = axis['device'].move(axis['value'])
-    while status.status == 'RUNNING':
-        for surv_ax in surveyed_axes:
-            if abs(surv_ax['device'].read() - surv_ax['old_value']) > surv_ax['abs_tol']:
-                axis['device'].stop()
-                raise MoveToLabelError(
-                    f"During movement of {axis['device'].name}, {surv_ax['device'].name} " +
-                    f"started to move unexpectedly (old pos: {surv_ax['old_value']}, " +
-                    f"current pos: {surv_ax['device'].read()})"
-                )

debye_bec/device_configs/x01da_experimental_hutch.yaml

@@ -1,3 +1,18 @@
+
+#######################################
+## Experimental Hutch Photon Shutter ##
+#######################################
+
+eh-sh:
+  readoutPriority: baseline
+  description: Experimental Hutch Photon Shutter
+  deviceClass: debye_bec.devices.eh_shutter.EHPhotonShutter
+  deviceConfig:
+    prefix: "X01DA-"
+  onFailure: retry
+  enabled: true
+  softwareTrigger: false
+
 ###################################
 ##         Optical Table         ##
 ###################################

debye_bec/device_configs/x01da_frontend.yaml

@@ -240,4 +240,4 @@ cm_xstripe:
     prefix: X01DA-FE-CM:XSTRIPE
   onFailure: retry
   enabled: true
-  softwareTrigger: false
+  softwareTrigger: false

debye_bec/device_configs/x01da_machine.yaml

@@ -15,4 +15,18 @@ curr:
   onFailure: buffer
   enabled: true
   readOnly: true
+  softwareTrigger: false
+
+bl_status:
+  readoutPriority: baseline
+  description: BL status for machine
+  deviceClass: ophyd.EpicsSignal
+  deviceConfig:
+    auto_monitor: false
+    read_pv: AGEOP-BL:STATUS-X01DA
+  deviceTags:
+    - machine
+  onFailure: buffer
+  enabled: true
+  readOnly: false
   softwareTrigger: false

debye_bec/device_configs/x01da_optics.yaml

@@ -1,4 +1,18 @@
 
+###################################
+##     Optics Photon Shutter     ##
+###################################
+
+op-sh:
+  readoutPriority: baseline
+  description: Optics Hutch Photon Shutter
+  deviceClass: debye_bec.devices.op_shutter.OPPhotonShutter
+  deviceConfig:
+    prefix: "X01DA-"
+  onFailure: retry
+  enabled: true
+  softwareTrigger: false
+
 ###################################
 ##         Monochromator         ##
 ###################################
@@ -408,4 +422,4 @@ sl2_gapy:
   softwareTrigger: false
   deviceTags:
     - optics
-    - slits
+    - slits

debye_bec/devices/absorber.py

@@ -10,9 +10,12 @@ from ophyd import EpicsSignal, EpicsSignalRO
 from ophyd_devices import CompareStatus, DeviceStatus
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
 
+from .utils.bl_status_enum import BlStatus
+
 if TYPE_CHECKING:
     from bec_lib.devicemanager import ScanInfo
 
+TIMEOUT_FOR_PV = 5
 
 class AbsorberError(Exception):
     """Absorber specific exception"""
@@ -37,6 +40,12 @@ class STATUS(int, enum.Enum):
     MAN_OPEN = 13
     UNDEFINED = 14
 
+class BL_ENABLE(int, enum.Enum):
+    """Beamline enable"""
+
+    DISABLE = 0
+    ENABLE = 1
+
 
 class Absorber(PSIDeviceBase):
     """Class for the Frontend Absorber"""
@@ -55,6 +64,7 @@ class Absorber(PSIDeviceBase):
         string=True,
         doc="Absorber Status",
     )
+    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)
@@ -63,12 +73,25 @@ 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) -> DeviceStatus | None:
-        """Open the Absorber"""
+    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 self.status.get() == STATUS.CLOSED:
+            if force:
+                if self.device_manager.bl_status.get() == BlStatus.OFFLINE:
+                    status = self.device_manager.bl_status.put(BlStatus.ATTENDED)
+                    status.wait(timeout=TIMEOUT_FOR_PV)
+                if self.close4bl.get() == BL_ENABLE.DISABLE:
+                    status = self.close4bl.set(BL_ENABLE.ENABLE)
+                    status.wait(timeout=TIMEOUT_FOR_PV)
             self.request.put(1)
-            status_open = CompareStatus(self.status, STATUS.OPEN, timeout=self.timeout_for_move)
-            status = status_open
+            status = CompareStatus(self.status, STATUS.OPEN, timeout=self.timeout_for_move)
             return status
         else:
             return None
@@ -77,8 +100,7 @@ class Absorber(PSIDeviceBase):
         """Close the Absorber"""
         if self.status.get() == STATUS.OPEN:
             self.request.put(1)
-            status_close = CompareStatus(self.status, STATUS.CLOSED, timeout=self.timeout_for_move)
-            status = status_close
+            status = CompareStatus(self.status, STATUS.CLOSED, timeout=self.timeout_for_move)
             return status
         else:
             return None

debye_bec/devices/eh_shutter.py

@@ -0,0 +1,75 @@
+"""Experimental Hutch Photon Shutter"""
+
+from __future__ import annotations
+
+import enum
+from typing import TYPE_CHECKING
+
+from ophyd import Component as Cpt
+from ophyd import EpicsSignal, EpicsSignalRO
+from ophyd_devices import CompareStatus, DeviceStatus
+from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+
+from .utils.bl_status_enum import BlStatus
+
+if TYPE_CHECKING:
+    from bec_lib.devicemanager import ScanInfo
+
+TIMEOUT_FOR_PV = 5
+
+class STATUS(int, enum.Enum):
+    """Shutter States"""
+
+    NOT_CLOSED = 0
+    CLOSED = 1
+
+class BL_ENABLE(int, enum.Enum):
+    """Beamline enable"""
+
+    DISABLE = 0
+    ENABLE = 1
+
+
+class OPPhotonShutter(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")
+    status = Cpt(
+        EpicsSignalRO, suffix="EH1", kind="normal", auto_monitor=True, doc="Shutter Status"
+    )
+    status_string = Cpt(
+        EpicsSignalRO,
+        suffix="EH1-PSYS:SH-A-CLOSE",
+        kind="normal",
+        auto_monitor=True,
+        string=True,
+        doc="Shutter Status",
+    )
+
+    def __init__(self, *, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs):
+        super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
+
+        self.timeout_for_move = 10
+        # Wait for connection on all components, ensure IOC is connected
+        self.wait_for_connection(all_signals=True, timeout=5)
+
+    def open(self) -> DeviceStatus | None:
+        """Open the Shutter"""
+        if self.status.get() == STATUS.CLOSED:
+            self.request_open.put(1)
+            status = CompareStatus(self.status, STATUS.NOT_CLOSED, timeout=self.timeout_for_move)
+            return status
+        else:
+            return None
+
+    def close(self) -> DeviceStatus | None:
+        """Close the Shutter"""
+        if self.status.get() == STATUS.NOT_CLOSED:
+            self.request_close.put(1)
+            status = CompareStatus(self.status, STATUS.CLOSED, timeout=self.timeout_for_move)
+            return status
+        else:
+            return None

debye_bec/devices/op_shutter.py

@@ -0,0 +1,94 @@
+"""Optics Photon Shutter"""
+
+from __future__ import annotations
+
+import enum
+from typing import TYPE_CHECKING
+
+from ophyd import Component as Cpt
+from ophyd import EpicsSignal, EpicsSignalRO
+from ophyd_devices import CompareStatus, DeviceStatus
+from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
+
+from .utils.bl_status_enum import BlStatus
+
+if TYPE_CHECKING:
+    from bec_lib.devicemanager import ScanInfo
+
+TIMEOUT_FOR_PV = 5
+
+class OPPhotonShutterError(Exception):
+    """Shutter specific exception"""
+
+
+class STATUS(int, enum.Enum):
+    """Shutter States"""
+
+    NOT_CLOSED = 0
+    CLOSED = 1
+
+class BL_ENABLE(int, enum.Enum):
+    """Beamline enable"""
+
+    DISABLE = 0
+    ENABLE = 1
+
+
+class OPPhotonShutter(PSIDeviceBase):
+    """Class for the Optics Photon Shutter"""
+
+    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")
+    status = Cpt(
+        EpicsSignalRO, suffix="OP-PSYS:SH-A-CLOSE", kind="normal", auto_monitor=True, doc="Shutter Status"
+    )
+    status_string = Cpt(
+        EpicsSignalRO,
+        suffix="OP-PSYS:SH-A-CLOSE",
+        kind="normal",
+        auto_monitor=True,
+        string=True,
+        doc="Shutter Status",
+    )
+    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)
+
+        self.timeout_for_move = 10
+        # 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:
+        """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 self.status.get() == STATUS.CLOSED:
+            if force:
+                if self.device_manager.bl_status.get() == BlStatus.OFFLINE:
+                    status = self.device_manager.bl_status.put(BlStatus.ATTENDED)
+                    status.wait(timeout=TIMEOUT_FOR_PV)
+                if self.close4bl.get() == BL_ENABLE.DISABLE:
+                    status = self.close4bl.set(BL_ENABLE.ENABLE)
+                    status.wait(timeout=TIMEOUT_FOR_PV)
+            self.request_open.put(1)
+            status = CompareStatus(self.status, STATUS.NOT_CLOSED, timeout=self.timeout_for_move)
+            return status
+        else:
+            return None
+
+    def close(self) -> DeviceStatus | None:
+        """Close the Shutter"""
+        if self.status.get() == STATUS.NOT_CLOSED:
+            self.request_close.put(1)
+            status = CompareStatus(self.status, STATUS.CLOSED, timeout=self.timeout_for_move)
+            return status
+        else:
+            return None

debye_bec/devices/utils/bl_status_enum.py

@@ -0,0 +1,8 @@
+import enum
+
+class BlStatus(str, enum.Enum):
+    """Beamline status enum"""
+
+    OFFLINE = 0
+    ATTENDED = 1
+    REMOTE = 2