diff --git a/debye_bec/bec_ipython_client/plugins/auto_gain.py b/debye_bec/bec_ipython_client/plugins/auto_gain.py new file mode 100644 index 0000000..ab4bbaa --- /dev/null +++ b/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 diff --git a/debye_bec/bec_ipython_client/plugins/move_to_label.py b/debye_bec/bec_ipython_client/plugins/move_to_label.py deleted file mode 100644 index dd14970..0000000 --- a/debye_bec/bec_ipython_client/plugins/move_to_label.py +++ /dev/null @@ -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()})" - ) diff --git a/debye_bec/device_configs/x01da_experimental_hutch.yaml b/debye_bec/device_configs/x01da_experimental_hutch.yaml index f2c241e..219e0d0 100644 --- a/debye_bec/device_configs/x01da_experimental_hutch.yaml +++ b/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 ## ################################### diff --git a/debye_bec/device_configs/x01da_frontend.yaml b/debye_bec/device_configs/x01da_frontend.yaml index 3a9edb7..77a4236 100644 --- a/debye_bec/device_configs/x01da_frontend.yaml +++ b/debye_bec/device_configs/x01da_frontend.yaml @@ -240,4 +240,4 @@ cm_xstripe: prefix: X01DA-FE-CM:XSTRIPE onFailure: retry enabled: true - softwareTrigger: false \ No newline at end of file + softwareTrigger: false diff --git a/debye_bec/device_configs/x01da_machine.yaml b/debye_bec/device_configs/x01da_machine.yaml index cd957dd..cbade1e 100644 --- a/debye_bec/device_configs/x01da_machine.yaml +++ b/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 \ No newline at end of file diff --git a/debye_bec/device_configs/x01da_optics.yaml b/debye_bec/device_configs/x01da_optics.yaml index a168e92..ea61ae9 100644 --- a/debye_bec/device_configs/x01da_optics.yaml +++ b/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 \ No newline at end of file + - slits diff --git a/debye_bec/devices/absorber.py b/debye_bec/devices/absorber.py index 73da60a..fa719db 100644 --- a/debye_bec/devices/absorber.py +++ b/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 diff --git a/debye_bec/devices/eh_shutter.py b/debye_bec/devices/eh_shutter.py new file mode 100644 index 0000000..52d06e3 --- /dev/null +++ b/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 diff --git a/debye_bec/devices/op_shutter.py b/debye_bec/devices/op_shutter.py new file mode 100644 index 0000000..99b8ba1 --- /dev/null +++ b/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 diff --git a/debye_bec/devices/utils/bl_status_enum.py b/debye_bec/devices/utils/bl_status_enum.py new file mode 100644 index 0000000..2913745 --- /dev/null +++ b/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