From 6189329168608f384b2e6ed8e78e6b32b508de6b Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Wed, 30 Sep 2026 16:10:25 +0200 Subject: [PATCH] chore: disable legacy scan We keep it in the repo for a later migration to v4 scans --- superxas_bec/scans/__init__.py | 8 +- superxas_bec/scans/exafs_scan.py | 226 +++++++++++++++---------------- 2 files changed, 115 insertions(+), 119 deletions(-) diff --git a/superxas_bec/scans/__init__.py b/superxas_bec/scans/__init__.py index f92a72b..5f4a83b 100644 --- a/superxas_bec/scans/__init__.py +++ b/superxas_bec/scans/__init__.py @@ -1,9 +1,5 @@ -from .exafs_scan import EXAFSScan - -from .mono_bragg_scans import ( +from .mono_bragg_scans import ( # XASAdvancedScanWithXRD,; XASSimpleScanWithXRD, XASAdvancedScan, - # XASAdvancedScanWithXRD, XASSimpleScan, - # XASSimpleScanWithXRD, ) -from .nidaq_cont_scan import NIDAQContinuousScan \ No newline at end of file +from .nidaq_cont_scan import NIDAQContinuousScan diff --git a/superxas_bec/scans/exafs_scan.py b/superxas_bec/scans/exafs_scan.py index a73e0f2..f81491f 100644 --- a/superxas_bec/scans/exafs_scan.py +++ b/superxas_bec/scans/exafs_scan.py @@ -1,131 +1,131 @@ -import time +# import time -import numpy as np -from bec_lib.device import DeviceBase -from bec_lib.logger import bec_logger -from bec_server.scan_server.scans import ScanBase +# import numpy as np +# from bec_lib.device import DeviceBase +# from bec_lib.logger import bec_logger +# from bec_server.scan_server.scans import ScanBase -logger = bec_logger.logger +# logger = bec_logger.logger -class EXAFSScan(ScanBase): - scan_name = "exafs_scan" +# class EXAFSScan(ScanBase): +# scan_name = "exafs_scan" - def __init__( - self, - edge_energy: float, - xas_rel_range: list[float] | np.ndarray[float] | None = None, - n_points: list[int] | np.ndarray[int] | None = None, - k_step: list[bool] | np.ndarray[bool] | None = None, - integ_time: list[float] | np.ndarray[float] | None = None, - motor: DeviceBase | None = None, - settling_time: float = 0.2, - **kwargs, - ): - """ - EXAFS Scan of the mono_energy axix +# def __init__( +# self, +# edge_energy: float, +# xas_rel_range: list[float] | np.ndarray[float] | None = None, +# n_points: list[int] | np.ndarray[int] | None = None, +# k_step: list[bool] | np.ndarray[bool] | None = None, +# integ_time: list[float] | np.ndarray[float] | None = None, +# motor: DeviceBase | None = None, +# settling_time: float = 0.2, +# **kwargs, +# ): +# """ +# EXAFS Scan of the mono_energy axix - Args: - edge_energy (float) : Adsorption Edge Energy - xas_rel_range (list[float] | np.ndarray[int] | None) : Optinoal, relative range for XAS, Length of list must n_points +1 - n_points (list[int] | np.ndarray[bool] | None) : Optional, number of points per range - ...#TODO docstring - """ - self.edge_energy = edge_energy - self.xas_rel_range = xas_rel_range - self.n_points = n_points - self.k_step = k_step - self.integ_time = integ_time - self.k_step_conversion = 3.81 - self._check_and_upated_input_arguments() - if motor is None: - default_motor = "kb_slit_y" - motor = default_motor # TODO Remove that motor, put energy of mono - self.motor = motor - super().__init__(exp_time=0, relative=False, settling_time=settling_time, **kwargs) +# Args: +# edge_energy (float) : Adsorption Edge Energy +# xas_rel_range (list[float] | np.ndarray[int] | None) : Optinoal, relative range for XAS, Length of list must n_points +1 +# n_points (list[int] | np.ndarray[bool] | None) : Optional, number of points per range +# ...#TODO docstring +# """ +# self.edge_energy = edge_energy +# self.xas_rel_range = xas_rel_range +# self.n_points = n_points +# self.k_step = k_step +# self.integ_time = integ_time +# self.k_step_conversion = 3.81 +# self._check_and_upated_input_arguments() +# if motor is None: +# default_motor = "kb_slit_y" +# motor = default_motor # TODO Remove that motor, put energy of mono +# self.motor = motor +# super().__init__(exp_time=0, relative=False, settling_time=settling_time, **kwargs) - # Check that trigger device is enabled - _dev_trigger_name = "trigger" - self._dev_trigger: DeviceBase = self.device_manager.devices.get(_dev_trigger_name, None) - if self._dev_trigger is None or self._dev_trigger.enabled == False: - raise ValueError( - f"Trigger device not found or not enabled in devicemanager {self._dev_trigger}" - ) - # update scan parameters - self.scan_parameters["edge_energy"] = self.edge_energy - self.scan_parameters["xas_rel_range"] = self.xas_rel_range - self.scan_parameters["n_points"] = self.n_points - self.scan_parameters["k_step"] = self.k_step - self.scan_parameters["integ_time"] = self.integ_time - # update readout_priority - self.readout_priority = {"monitored": [self.motor]} +# # Check that trigger device is enabled +# _dev_trigger_name = "trigger" +# self._dev_trigger: DeviceBase = self.device_manager.devices.get(_dev_trigger_name, None) +# if self._dev_trigger is None or self._dev_trigger.enabled == False: +# raise ValueError( +# f"Trigger device not found or not enabled in devicemanager {self._dev_trigger}" +# ) +# # update scan parameters +# self.scan_parameters["edge_energy"] = self.edge_energy +# self.scan_parameters["xas_rel_range"] = self.xas_rel_range +# self.scan_parameters["n_points"] = self.n_points +# self.scan_parameters["k_step"] = self.k_step +# self.scan_parameters["integ_time"] = self.integ_time +# # update readout_priority +# self.readout_priority = {"monitored": [self.motor]} - def update_scan_motors(self): - self.scan_motors = [self.motor] +# def update_scan_motors(self): +# self.scan_motors = [self.motor] - def _check_and_upated_input_arguments(self) -> None: - """ - Input parameters for the EXAFS scan must be of the same length for n_points, k_step, integ_time - and xas_rel_range (-1). This methods checks for this condition, and calculates the integration time - for each point in the scan. If the input parameters are not provided with the correct length, - it raises a ValueError which indicates which parameters are not matching. - """ +# def _check_and_upated_input_arguments(self) -> None: +# """ +# Input parameters for the EXAFS scan must be of the same length for n_points, k_step, integ_time +# and xas_rel_range (-1). This methods checks for this condition, and calculates the integration time +# for each point in the scan. If the input parameters are not provided with the correct length, +# it raises a ValueError which indicates which parameters are not matching. +# """ - if not all( - [ - len(self.n_points) == len(self.k_step), - len(self.n_points) == len(self.integ_time), - len(self.n_points) == (len(self.xas_rel_range) - 1), # careful -1 - ] - ): - raise ValueError( - f"Input parameters must have matching lengths:\n" - f"n_points: {len(self.n_points)}, " - f"k_step: {len(self.k_step)}, expected: {len(self.n_points)}, " - f"integ_time: {len(self.integ_time)}, expected: {len(self.n_points)}, " - f"xas_rel_range: {len(self.xas_rel_range)}, expected: {len(self.n_points)-1} " - ) +# if not all( +# [ +# len(self.n_points) == len(self.k_step), +# len(self.n_points) == len(self.integ_time), +# len(self.n_points) == (len(self.xas_rel_range) - 1), # careful -1 +# ] +# ): +# raise ValueError( +# f"Input parameters must have matching lengths:\n" +# f"n_points: {len(self.n_points)}, " +# f"k_step: {len(self.k_step)}, expected: {len(self.n_points)}, " +# f"integ_time: {len(self.integ_time)}, expected: {len(self.n_points)}, " +# f"xas_rel_range: {len(self.xas_rel_range)}, expected: {len(self.n_points)-1} " +# ) - self.integ_time = np.repeat(np.array(self.integ_time), np.array(self.n_points)) +# self.integ_time = np.repeat(np.array(self.integ_time), np.array(self.n_points)) - def _set_position_offset(self): - """Do not set offset""" - yield None +# def _set_position_offset(self): +# """Do not set offset""" +# yield None - def _calculate_positions(self): - positions = [] - for ii, pnts in enumerate(self.n_points): - if self.k_step[ii] is False: - positions.extend( - np.linspace( - self.xas_rel_range[ii], self.xas_rel_range[ii + 1], pnts, endpoint=False - ).tolist() - ) - else: - k_start = np.sqrt(self.xas_rel_range[ii] / self.k_step_conversion) - k_stop = np.sqrt(self.xas_rel_range[ii + 1] / self.k_step_conversion) - k_pos = np.linspace(k_start, k_stop, pnts, endpoint=False) - k_pos = k_pos**2 * self.k_step_conversion - positions.extend(k_pos.tolist()) +# def _calculate_positions(self): +# positions = [] +# for ii, pnts in enumerate(self.n_points): +# if self.k_step[ii] is False: +# positions.extend( +# np.linspace( +# self.xas_rel_range[ii], self.xas_rel_range[ii + 1], pnts, endpoint=False +# ).tolist() +# ) +# else: +# k_start = np.sqrt(self.xas_rel_range[ii] / self.k_step_conversion) +# k_stop = np.sqrt(self.xas_rel_range[ii + 1] / self.k_step_conversion) +# k_pos = np.linspace(k_start, k_stop, pnts, endpoint=False) +# k_pos = k_pos**2 * self.k_step_conversion +# positions.extend(k_pos.tolist()) - # Create positions array - self.positions = np.vstack(positions) - # shift by edge energy - self.positions = self.positions + self.edge_energy - # Convert to keV - self.positions = self.positions / 1e3 +# # Create positions array +# self.positions = np.vstack(positions) +# # shift by edge energy +# self.positions = self.positions + self.edge_energy +# # Convert to keV +# self.positions = self.positions / 1e3 - def _at_each_point(self, ind=None, pos=None): - yield from self._move_scan_motors_and_wait(pos) - time.sleep(self.settling_time) - trigger_time = self.integ_time[ind] - self.stubs.send_rpc_and_wait(self._dev_trigger, "set_exposure_time", trigger_time) +# def _at_each_point(self, ind=None, pos=None): +# yield from self._move_scan_motors_and_wait(pos) +# time.sleep(self.settling_time) +# trigger_time = self.integ_time[ind] +# self.stubs.send_rpc_and_wait(self._dev_trigger, "set_exposure_time", trigger_time) - # Trigger - yield from self.stubs.trigger(min_wait=trigger_time) +# # Trigger +# yield from self.stubs.trigger(min_wait=trigger_time) - # Readout all monitored devices - yield from self.stubs.read(group="monitored", point_id=self.point_id) +# # Readout all monitored devices +# yield from self.stubs.read(group="monitored", point_id=self.point_id) - # Increase point id - self.point_id += 1 +# # Increase point id +# self.point_id += 1 -- 2.54.0