From cd13e7ed9e4c88e6b9b18bbdf371df81e1a4ddd4 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Wed, 20 May 2026 10:23:17 +0200 Subject: [PATCH] feat: migrate to v4 scans --- .../bec_ipython_client/plugins/LamNI/lamni.py | 55 +- .../Endstation INFO flomni lamni omny.md | 7 +- .../plugins/flomni/flomni.py | 4 +- .../widgets/tomo_params/tomo_params.py | 43 +- csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py | 3 - csaxs_bec/devices/sim/sim_lamni.py | 2 +- csaxs_bec/scans/LamNIFermatScan.py | 554 ------------------ csaxs_bec/scans/__init__.py | 5 +- csaxs_bec/scans/flomni_fermat_scan.py | 529 +++++++++-------- csaxs_bec/scans/lamni_components.py | 179 ++++++ csaxs_bec/scans/lamni_fermat_scan.py | 525 +++++++++++++++++ csaxs_bec/scans/lamni_move_to_scan_center.py | 115 ++++ csaxs_bec/scans/omny_fermat_scan.py | 465 ++++++++------- .../test_fermat_position_warning.py | 15 +- .../test_lamni_tomo_params_widget_math.py | 11 +- .../test_tomo_params_widget_math.py | 4 +- tests/tests_scans/test_flomni_fermat_scan.py | 57 -- tests/tests_scans/test_lamni_fermat_scan.py | 422 ------------- 18 files changed, 1456 insertions(+), 1539 deletions(-) delete mode 100644 csaxs_bec/scans/LamNIFermatScan.py create mode 100644 csaxs_bec/scans/lamni_components.py create mode 100644 csaxs_bec/scans/lamni_fermat_scan.py create mode 100644 csaxs_bec/scans/lamni_move_to_scan_center.py delete mode 100644 tests/tests_scans/test_flomni_fermat_scan.py delete mode 100644 tests/tests_scans/test_lamni_fermat_scan.py diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py index 81af541..fbf34f5 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py @@ -12,17 +12,17 @@ from bec_lib.pdf_writer import PDFWriter from bec_lib.scan_repeat import scan_repeat from typeguard import typechecked +from csaxs_bec.bec_ipython_client.plugins.LamNI.gui_tools import LamniGuiTools +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_alignment_mixin import LamNIAlignmentMixin from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools import ( OMNYTools, PtychoReconstructor, TomoIDManager, ) from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.tomo_queue_mixin import TomoQueueMixin -from csaxs_bec.bec_ipython_client.plugins.LamNI.gui_tools import LamniGuiTools -from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_alignment_mixin import LamNIAlignmentMixin -from .x_ray_eye_align import XrayEyeAlign as XrayEyeAlignGUI from .lamni_optics_mixin import LaMNIInitStages, LamNIOpticsMixin +from .x_ray_eye_align import XrayEyeAlign as XrayEyeAlignGUI logger = bec_logger.logger @@ -885,9 +885,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools @golden_projections_at_0_deg_for_damage_estimation.setter def golden_projections_at_0_deg_for_damage_estimation(self, val: int): - self.client.set_global_var( - "golden_projections_at_0_deg_for_damage_estimation", val - ) + self.client.set_global_var("golden_projections_at_0_deg_for_damage_estimation", val) @property def zero_deg_reference_at_each_subtomo(self): @@ -1179,7 +1177,8 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools ) corridor_size = self.corridor_size if self.corridor_size > 0 else None scans.lamni_fermat_scan( - fov_size=[self.lamni_piezo_range_x, self.lamni_piezo_range_y], + fovx=self.lamni_piezo_range_x, + fovy=self.lamni_piezo_range_y, step=self.tomo_shellstep, stitch_x=stitch_x, stitch_y=stitch_y, @@ -1200,10 +1199,9 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools ), fov_circular=self.tomo_circfov, angle=angle, - scan_type="fly", exp_time=self.tomo_countingtime, frames_per_trigger=self.frames_per_trigger, - optim_trajectory_corridor=corridor_size, + corridor_size=corridor_size, ) def write_alignment_scan_numbers(self, first_scan: int) -> None: @@ -1271,7 +1269,9 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools alignment_scan_numbers = [] self.alignment_scan_progress.reset() - self.alignment_scan_progress.update(total_angles=len(angles), angle_index=0, angle=angles[0]) + self.alignment_scan_progress.update( + total_angles=len(angles), angle_index=0, angle=angles[0] + ) self.lamnigui_show_alignment_progress() for idx, angle in enumerate(angles): @@ -1605,9 +1605,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools angular_step = maxangle / number_of_projections_per_subtomo subtomo_number = int((ii * angular_step) / maxangle) + 1 start_angle = self._golden(subtomo_number - 1, 1, angular_step)[0] - projection_number_of_subtomo = ( - ii - (subtomo_number - 1) * number_of_projections_per_subtomo - ) + projection_number_of_subtomo = ii - (subtomo_number - 1) * number_of_projections_per_subtomo if reverse: if subtomo_number % 2: @@ -1905,8 +1903,8 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools def _expected_fermat_position_count(self) -> int: """Predict the number of Fermat-spiral scan positions the current settings would produce per projection tile, using the exact same - algorithm LamNIFermatScan runs at scan time - (LamNIFermatScan.get_lamni_fermat_spiral_pos()) -- so a + algorithm LamniFermatScan runs at scan time + (LamniFermatScan.get_lamni_fermat_spiral_pos()) -- so a too-few-points configuration (which the scan server would only catch by aborting with ScanAbortion once the scan actually starts) can be caught here instead, while just looking at tomo_parameters(). @@ -1920,9 +1918,9 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools tile, since the circular FOV crop (tomo_circfov) is checked against the rotated stage position. """ - from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan + from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan - positions = LamNIFermatScan.get_lamni_fermat_spiral_pos( + positions = LamniFermatScan.get_lamni_fermat_spiral_pos( -abs(self.lamni_piezo_range_x / 2), abs(self.lamni_piezo_range_x / 2), -abs(self.lamni_piezo_range_y / 2), @@ -1934,18 +1932,19 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools stitch_x=0, stitch_y=0, stitch_overlap=self.tomo_stitch_overlap, - fov_size=[self.lamni_piezo_range_x, self.lamni_piezo_range_y], + fovx=self.lamni_piezo_range_x, + fovy=self.lamni_piezo_range_y, fov_circular=self.tomo_circfov, ) return len(positions) @staticmethod def _fermat_min_positions() -> int: - """LamNIFermatScan's own minimum-position threshold -- see + """LamniFermatScan's own minimum-position threshold -- see _expected_fermat_position_count().""" - from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan + from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan - return LamNIFermatScan._MIN_POSITIONS + return LamniFermatScan.MIN_POSITIONS def tomo_parameters(self): """Print and interactively update the tomo parameters.""" @@ -2011,7 +2010,6 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools print("Repeating projections at 0 deg at start of every second subtomogram.") print(f"\nSample name: {self.sample_name}\n") - if self.OMNYTools.yesno("Are these parameters correctly set for your scan?", "y"): print("OK. continue.") return @@ -2091,9 +2089,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools elif self.tomo_type == 3: numprj = self._get_val( - "Number of projections per sub-tomogram", - int(360 / self.tomo_angle_stepsize), - int, + "Number of projections per sub-tomogram", int(360 / self.tomo_angle_stepsize), int ) self.tomo_angle_stepsize = 360 / numprj self.golden_max_number_of_projections = self._get_val( @@ -2234,7 +2230,9 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools content.append(f"{'At-each-angle hook:':<{padding}}{hook_description}\n") content = "".join(content) hook_source = self._active_hook_source() - user_target = os.path.expanduser(f"~/data/raw/documentation/tomo_scan_ID_{self.tomo_id}.pdf") + user_target = os.path.expanduser( + f"~/data/raw/documentation/tomo_scan_ID_{self.tomo_id}.pdf" + ) with PDFWriter(user_target) as file: self._add_psi_footer(file) # PDFWriter (bec_lib) has no public image API -- reach into its @@ -2271,6 +2269,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools def get_calibration_of_capstops_left_and_right(self): import time + print(""" Manual on how to center the Piezo stage first. To obtain the center voltages one can move in closed loop to the interferometer @@ -2290,7 +2289,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools voltage2 = float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[2]")) if angle < 360: print(f"{angle},{voltage1},{voltage2}") - time.sleep(.3) + time.sleep(0.3) time.sleep(10) print("\nCapstop left\nAngle, Voltage1, Voltage2") @@ -2301,6 +2300,6 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools voltage2 = float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[2]")) if angle > 0: print(f"{angle},{voltage1},{voltage2}") - time.sleep(.3) + time.sleep(0.3) print("Finished") diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/Endstation INFO flomni lamni omny.md b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/Endstation INFO flomni lamni omny.md index 097d940..77f5611 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/Endstation INFO flomni lamni omny.md +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/Endstation INFO flomni lamni omny.md @@ -163,8 +163,8 @@ Prozedur bei Rotationswechsel: ```python # Standard LamNI Fermat-Scan scans.lamni_fermat_scan( - fov_size=[20], # FOV in Piezo-Ebene [µm], max ~80 µm - # [x] = quadratisch, [x,y] = rechteckig + fovx=20, # FOV X in Piezo-Ebene [µm], max ~80 µm + fovy=20, # FOV Y in Piezo-Ebene [µm], max ~80 µm step=0.5, # Schrittweite [µm] exp_time=0.1, # Belichtungszeit [s] angle=0, # Laminographie-Rotationswinkel [Grad] @@ -176,7 +176,6 @@ scans.lamni_fermat_scan( stitch_y=0, # Stitch-Versatz Y [µm] fov_circular=0, # Kreisförmiges FOV [µm] (zusätzliches Cropping) stitch_overlap=1, # Stitch-Überlapp [µm] - scan_type="fly", # "fly" (HW-getriggert) oder "step" frames_per_trigger=1, ) @@ -604,4 +603,4 @@ BEC-Treiber: `csaxs_bec/devices/omny/galil/` 16. **OMNY Tracking-Stage y-Piezo** (ZSA-400-PSI, 400 µm, 200 Hz): ähnliche Eigenschaften wie Delta-Scanner – daher geeignet für kontinuierliche Bewegung während Ptychographie --- -*Erstellt April 2026. Quellen: Quellcode csaxs_bec + 4 peer-reviewed Publikationen (Holler et al. 2012, 2015, 2018, 2020).* \ No newline at end of file +*Erstellt April 2026. Quellen: Quellcode csaxs_bec + 4 peer-reviewed Publikationen (Holler et al. 2012, 2015, 2018, 2020).* diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py index 34bb8ba..c3f228f 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py @@ -3426,7 +3426,7 @@ class Flomni( _expected_fermat_position_count().""" from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan - return FlomniFermatScan._MIN_POSITIONS + return FlomniFermatScan.MIN_POSITIONS def tomo_parameters(self): """print and update the tomo parameters""" @@ -3913,4 +3913,4 @@ if __name__ == "__main__": builtins.__dict__["bec"] = bec builtins.__dict__["umv"] = umv flomni = Flomni(bec) - flomni.start_x_ray_eye_alignment() \ No newline at end of file + flomni.start_x_ray_eye_alignment() diff --git a/csaxs_bec/bec_widgets/widgets/tomo_params/tomo_params.py b/csaxs_bec/bec_widgets/widgets/tomo_params/tomo_params.py index f58fe23..0cc6230 100644 --- a/csaxs_bec/bec_widgets/widgets/tomo_params/tomo_params.py +++ b/csaxs_bec/bec_widgets/widgets/tomo_params/tomo_params.py @@ -93,7 +93,10 @@ TOMO_TYPES = { # job), on both flomni and lamni. _TYPE1_ONLY_PARAMS = {"tomo_angle_range", "zero_deg_reference_at_each_subtomo"} _TYPE2_ONLY_PARAMS = {"golden_ratio_bunch_size"} -_TYPE23_PARAMS = {"golden_max_number_of_projections", "golden_projections_at_0_deg_for_damage_estimation"} +_TYPE23_PARAMS = { + "golden_max_number_of_projections", + "golden_projections_at_0_deg_for_damage_estimation", +} def _irrelevant_params_for_type(tomo_type: Any) -> set[str]: @@ -108,6 +111,7 @@ def _irrelevant_params_for_type(tomo_type: Any) -> set[str]: irrelevant |= _TYPE2_ONLY_PARAMS return irrelevant + STATUS_COLORS = { "pending": "#888888", "running": "#2196F3", @@ -1017,7 +1021,7 @@ class TomoParamsWidget(BECWidget, QWidget): if reason: banner.setText( f"\u26a0 Beamline busy — {reason}.\n" - "Editing is allowed, but Submit is blocked -- use \"Add to queue\" " + 'Editing is allowed, but Submit is blocked -- use "Add to queue" ' "to save edits as a new job instead." ) banner.setVisible(True) @@ -1073,8 +1077,7 @@ class TomoParamsWidget(BECWidget, QWidget): reply = QMessageBox.question( self, "Discard current edit?", - f"An edit is already in progress. Loading {what} will discard " - "it. Continue?", + f"An edit is already in progress. Loading {what} will discard " "it. Continue?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, ) if reply != QMessageBox.StandardButton.Yes: @@ -1147,7 +1150,7 @@ class TomoParamsWidget(BECWidget, QWidget): f"Can't submit while the beamline is busy ({busy_reason}) — " "these are the live scan parameters and writing them would " "perturb the running acquisition.\n\n" - "Use \"Add to queue\" instead to save these edits as a new " + 'Use "Add to queue" instead to save these edits as a new ' "queue job without touching the running scan.", ) return @@ -1231,9 +1234,7 @@ class TomoParamsWidget(BECWidget, QWidget): if reply != QMessageBox.StandardButton.Yes: return - label, ok = QInputDialog.getText( - self, "Add to queue", "Job label (leave blank for auto):" - ) + label, ok = QInputDialog.getText(self, "Add to queue", "Job label (leave blank for auto):") if not ok: return @@ -1435,7 +1436,7 @@ class TomoQueueDialog(QDialog): self._btn_load.setToolTip( "Load the selected tomo job's saved settings into the params panel's " "editor -- doesn't touch live params or the queue itself. Review, " - "tweak, then Submit or \"Add to queue\" from there." + 'tweak, then Submit or "Add to queue" from there.' ) self._btn_load.setEnabled(False) self._btn_del = QPushButton("Delete selected") @@ -1449,8 +1450,7 @@ class TomoQueueDialog(QDialog): self._btn_sort = QPushButton("Sort queue…") self._btn_sort.setCheckable(True) self._btn_sort.setToolTip( - "Reorder pending jobs. The running/incomplete job (if any) and " - "done jobs stay put." + "Reorder pending jobs. The running/incomplete job (if any) and " "done jobs stay put." ) self._btn_move_up = QPushButton("▲ Move up") self._btn_move_down = QPushButton("▼ Move down") @@ -1549,7 +1549,7 @@ class TomoQueueDialog(QDialog): "Unsaved edit in progress", "The tomo parameters panel has an edit in progress. This button " "queues the current LIVE parameters -- not your unsaved edit. Use " - "\"Add to queue\" in the params panel itself if you want to queue " + '"Add to queue" in the params panel itself if you want to queue ' "what you just typed.\n\n" "Queue the live (unedited) parameters anyway?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, @@ -2163,16 +2163,16 @@ def _compute_fermat_positions_flomni(params: dict[str, Any]) -> tuple[int, int]: fovy = params.get("fovy", 0.0) step = params.get("tomo_shellstep", 0.0) if step <= 0: - return 0, FlomniFermatScan._MIN_POSITIONS + return 0, FlomniFermatScan.MIN_POSITIONS positions = FlomniFermatScan.get_flomni_fermat_spiral_pos( -abs(fovx / 2), abs(fovx / 2), -abs(fovy / 2), abs(fovy / 2), step=step, spiral_type=0 ) - return len(positions), FlomniFermatScan._MIN_POSITIONS + return len(positions), FlomniFermatScan.MIN_POSITIONS def _compute_fermat_positions_lamni(params: dict[str, Any]) -> tuple[int, int]: """Lamni sibling of _compute_fermat_positions_flomni(): calls - LamNIFermatScan.get_lamni_fermat_spiral_pos() directly -- including its + LamniFermatScan.get_lamni_fermat_spiral_pos() directly -- including its rotated-stage and circular-FOV (tomo_circfov) cropping, which can matter a lot and would be easy to get subtly wrong in a reimplementation. Evaluated at the *currently edited* stitch tile and angle=0 (lamni has @@ -2182,14 +2182,14 @@ def _compute_fermat_positions_lamni(params: dict[str, Any]) -> tuple[int, int]: Returns: (estimated_count, minimum_required) """ - from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan + from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan piezo_x = params.get("lamni_piezo_range_x", 0.0) piezo_y = params.get("lamni_piezo_range_y", 0.0) step = params.get("tomo_shellstep", 0.0) if step <= 0: - return 0, LamNIFermatScan._MIN_POSITIONS - positions = LamNIFermatScan.get_lamni_fermat_spiral_pos( + return 0, LamniFermatScan.MIN_POSITIONS + positions = LamniFermatScan.get_lamni_fermat_spiral_pos( -abs(piezo_x / 2), abs(piezo_x / 2), -abs(piezo_y / 2), @@ -2200,10 +2200,11 @@ def _compute_fermat_positions_lamni(params: dict[str, Any]) -> tuple[int, int]: stitch_x=params.get("lamni_stitch_x", 0), stitch_y=params.get("lamni_stitch_y", 0), stitch_overlap=params.get("tomo_stitch_overlap", 1.0), - fov_size=[piezo_x, piezo_y], + fovx=piezo_x, + fovy=piezo_y, fov_circular=params.get("tomo_circfov", 0.0), ) - return len(positions), LamNIFermatScan._MIN_POSITIONS + return len(positions), LamniFermatScan.MIN_POSITIONS def _lamni_get_tomo_fov_offset(client, axis: str) -> float: @@ -2453,4 +2454,4 @@ def _fmt_num(val) -> str: def _color_from_hex(hex_color: str): from qtpy.QtGui import QColor - return QColor(hex_color) \ No newline at end of file + return QColor(hex_color) diff --git a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py index c482ceb..123aab5 100644 --- a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py +++ b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py @@ -796,9 +796,6 @@ class RtFlomniMotor(Device, PositionerBase): else: raise TypeError(f"Expected value of type int but received {type(val)}") - def kickoff(self, metadata, **kwargs) -> None: - self.controller.kickoff(metadata) - @property def egu(self): """The engineering units (EGU) for positions""" diff --git a/csaxs_bec/devices/sim/sim_lamni.py b/csaxs_bec/devices/sim/sim_lamni.py index 55c6be8..406c237 100644 --- a/csaxs_bec/devices/sim/sim_lamni.py +++ b/csaxs_bec/devices/sim/sim_lamni.py @@ -30,7 +30,7 @@ from csaxs_bec.devices.sim.sim_galil import SimGalilSocket, SimGalilState, _sim_ from csaxs_bec.devices.sim.sim_socket import SimSocketBase, SimStateRegistry # single source of truth for the LamNI geometry (same constants the scan uses) -from csaxs_bec.scans.LamNIFermatScan import MOVEMENT_SCALE_X, MOVEMENT_SCALE_Y +from csaxs_bec.scans.lamni_components import MOVEMENT_SCALE_X, MOVEMENT_SCALE_Y logger = bec_logger.logger diff --git a/csaxs_bec/scans/LamNIFermatScan.py b/csaxs_bec/scans/LamNIFermatScan.py deleted file mode 100644 index 83da5b1..0000000 --- a/csaxs_bec/scans/LamNIFermatScan.py +++ /dev/null @@ -1,554 +0,0 @@ -""" -SCAN PLUGINS - -All new scans should be derived from ScanBase. ScanBase provides various methods that can be customized and overriden -but they are executed in a specific order: - -- self.initialize # initialize the class if needed -- self.read_scan_motors # used to retrieve the start position (and the relative position shift if needed) -- self.prepare_positions # prepare the positions for the scan. The preparation is split into multiple sub fuctions: - - self._calculate_positions # calculate the positions - - self._set_positions_offset # apply the previously retrieved scan position shift (if needed) - - self._check_limits # tests to ensure the limits won't be reached -- self.open_scan # send an open_scan message including the scan name, the number of points and the scan motor names -- self.stage # stage all devices for the upcoming acquisiton -- self.run_baseline_readings # read all devices to get a baseline for the upcoming scan -- self.scan_core # run a loop over all position - - self._at_each_point(ind, pos) # called at each position with the current index and the target positions as arguments -- self.finalize # clean up the scan, e.g. move back to the start position; wait everything to finish -- self.unstage # unstage all devices that have been staged before -- self.cleanup # send a close scan message and perform additional cleanups if needed -""" - -import time - -import numpy as np -from bec_lib import bec_logger -from bec_lib.endpoints import MessageEndpoints -from bec_server.scan_server.errors import ScanAbortion -from bec_server.scan_server.scans import AsyncFlyScanBase, RequestBase, ScanArgType - -MOVEMENT_SCALE_X = np.sin(np.radians(15)) * np.cos(np.radians(30)) -MOVEMENT_SCALE_Y = np.cos(np.radians(15)) - -logger = bec_logger.logger - - -def lamni_to_stage_coordinates(x: float, y: float) -> tuple: - """convert from lamni coordinates to stage coordinates""" - y_stage = y / MOVEMENT_SCALE_Y - x_stage = 2 * (x - y_stage * MOVEMENT_SCALE_X) - return (x_stage, y_stage) - - -def lamni_from_stage_coordinates(x_stage: float, y_stage: float) -> tuple: - """convert to lamni coordinates from stage coordinates""" - x = x_stage * 0.5 + y_stage * MOVEMENT_SCALE_X - y = y_stage * MOVEMENT_SCALE_Y - return (x, y) - - -class LamNIMixin: - @staticmethod - def _lamni_compute_scan_center(x, y, angle_deg): - # assuming a scan point was found at interferometer x,y at zero degrees - # this function computes the new interferometer coordinates of this spot - # at a different rotation angle based on the lamni geometry - alpha = angle_deg / 180 * np.pi - stage_x, stage_y = lamni_to_stage_coordinates(x, y) - stage_x_rot = np.cos(alpha) * stage_x - np.sin(alpha) * stage_y - stage_y_rot = np.sin(alpha) * stage_x + np.cos(alpha) * stage_y - return lamni_from_stage_coordinates(stage_x_rot, stage_y_rot) - - def lamni_new_scan_center_interferometer(self, x, y): - """move to new scan center. xy in mm""" - lsamx_user_params = self.device_manager.devices.lsamx.user_parameter - if lsamx_user_params is None or lsamx_user_params.get("center") is None: - raise RuntimeError("lsamx center is not defined") - lsamy_user_params = self.device_manager.devices.lsamy.user_parameter - if lsamy_user_params is None or lsamy_user_params.get("center") is None: - raise RuntimeError("lsamy center is not defined") - lsamx_center = lsamx_user_params.get("center") - lsamy_center = lsamy_user_params.get("center") - - # could first check if feedback is enabled - yield from self.stubs.send_rpc_and_wait("rtx", "controller.feedback_disable") - time.sleep(0.05) - - rtx_current = yield from self.stubs.send_rpc_and_wait("rtx", "readback.get") - rty_current = yield from self.stubs.send_rpc_and_wait("rty", "readback.get") - lsamx_current = yield from self.stubs.send_rpc_and_wait("lsamx", "readback.get") - lsamy_current = yield from self.stubs.send_rpc_and_wait("lsamy", "readback.get") - - x_stage, y_stage = lamni_to_stage_coordinates(x, y) - - x_center_expect, y_center_expect = lamni_from_stage_coordinates( - lsamx_current - lsamx_center, lsamy_current - lsamy_center - ) - - # in microns - x_drift = x_center_expect * 1000 - rtx_current - y_drift = y_center_expect * 1000 - rty_current - - logger.info(f"Current uncompensated drift of setup is x={x_drift:.3f}, y={y_drift:.3f}") - - move_x = x_stage + lsamx_center + lamni_to_stage_coordinates(x_drift, y_drift)[0] / 1000 - move_y = y_stage + lsamy_center + lamni_to_stage_coordinates(x_drift, y_drift)[1] / 1000 - - coarse_move_req_x = np.abs(lsamx_current - move_x) - coarse_move_req_y = np.abs(lsamy_current - move_y) - - self.device_manager.devices.lsamx.read_only = False - self.device_manager.devices.lsamy.read_only = False - - if ( - np.abs(y_drift) > 150 - or np.abs(x_drift) > 150 - or (coarse_move_req_y < 0.003 and coarse_move_req_x < 0.003) - ): - logger.info("No drift correction.") - else: - logger.info( - f"Compensating {[val/1000 for val in lamni_to_stage_coordinates(x_drift,y_drift)]}" - ) - yield from self.stubs.set(device="lsamx", value=move_x) - yield from self.stubs.set(device="lsamy", value=move_y) - - time.sleep(0.01) - rtx_current = yield from self.stubs.send_rpc_and_wait("rtx", "readback.get") - rty_current = yield from self.stubs.send_rpc_and_wait("rty", "readback.get") - - logger.info(f"New scan center interferometer {rtx_current:.3f}, {rty_current:.3f} microns") - - # second iteration - x_center_expect, y_center_expect = lamni_from_stage_coordinates(x_stage, y_stage) - - # in microns - x_drift2 = x_center_expect * 1000 - rtx_current - y_drift2 = y_center_expect * 1000 - rty_current - logger.info( - f"Uncompensated drift of setup after first iteration is x={x_drift2:.3f}," - f" y={y_drift2:.3f}" - ) - - if np.abs(x_drift2) > 5 or np.abs(y_drift2) > 5: - logger.info( - "Compensating second iteration" - f" {[val/1000 for val in lamni_to_stage_coordinates(x_drift2,y_drift2)]}" - ) - move_x = ( - x_stage - + lsamx_center - + lamni_to_stage_coordinates(x_drift, y_drift)[0] / 1000 - + lamni_to_stage_coordinates(x_drift2, y_drift2)[0] / 1000 - ) - move_y = ( - y_stage - + lsamy_center - + lamni_to_stage_coordinates(x_drift, y_drift)[1] / 1000 - + lamni_to_stage_coordinates(x_drift2, y_drift2)[1] / 1000 - ) - yield from self.stubs.set(device="lsamx", value=move_x) - yield from self.stubs.set(device="lsamy", value=move_y) - - time.sleep(0.01) - rtx_current = yield from self.stubs.send_rpc_and_wait("rtx", "readback.get") - rty_current = yield from self.stubs.send_rpc_and_wait("rty", "readback.get") - - logger.info( - f"New scan center interferometer after second iteration {rtx_current:.3f}," - f" {rty_current:.3f} microns" - ) - x_drift2 = x_center_expect * 1000 - rtx_current - y_drift2 = y_center_expect * 1000 - rty_current - logger.info( - f"Uncompensated drift of setup after second iteration is x={x_drift2:.3f}," - f" y={y_drift2:.3f}" - ) - else: - logger.info("No second iteration required") - - self.device_manager.devices.lsamx.read_only = True - self.device_manager.devices.lsamy.read_only = True - - # update angle readback before start of the scan - yield from self.stubs.send_rpc_and_wait("lsamrot", "readback.get") - - yield from self.stubs.send_rpc_and_wait("rtx", "controller.feedback_enable_without_reset") - - -class LamNIMoveToScanCenter(RequestBase, LamNIMixin): - scan_name = "lamni_move_to_scan_center" - scan_report_hint = None - scan_type = "step" - required_kwargs = ["shift_x", "shift_y", "angle"] - arg_input = {} - arg_bundle_size = {"bundle": 0, "min": 0, "max": 0} - - def __init__(self, *args, parameter=None, **kwargs): - """ - Move LamNI to a new scan center. - - Args: - shift_x (float): shift x in mm - shift_y (float): shift y in mm - angle (float): tomo angle in degrees - - Examples: - >>> scans.lamni_move_to_scan_center(shift_x=1.2, shift_y=2.8, angle=12.5) - """ - super().__init__(parameter=parameter, **kwargs) - scan_kwargs = parameter.get("kwargs", {}) - self.shift_x = float(scan_kwargs.get("shift_x", 0)) - self.shift_y = float(scan_kwargs.get("shift_y", 0)) - self.angle = float(scan_kwargs.get("angle", 0)) - - def run(self): - center_x, center_y = self._lamni_compute_scan_center( - self.shift_x, self.shift_y, self.angle - ) - yield from self.lamni_new_scan_center_interferometer(center_x, center_y) - - -class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin): - scan_name = "lamni_fermat_scan" - scan_type = "fly" - required_kwargs = ["fov_size", "exp_time", "step", "angle"] - arg_input = {} - arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None} - - # Minimum number of Fermat-spiral positions a scan is allowed to run - # with -- exposed as a class attribute (not just a literal inside - # _check_min_positions()) so client-side code (tomo_parameters(), - # tomo_params.py) can warn about a too-few-points configuration before - # it ever reaches the scan server, using the exact same threshold. - _MIN_POSITIONS = 20 - - def __init__(self, *args, parameter: dict = None, frames_per_trigger:int=1, exp_time:float=0,**kwargs): - """ - A LamNI scan following Fermat's spiral. - - Kwargs: - fov_size [um]: Fov in the piezo plane (i.e. piezo range). Max 80 um - step [um]: stepsize - shift_x/y [mm]: extra shift in x/y. The shift is directly applied to the scan. It will not be auto rotated. (default 0). - center_x/center_y [mm]: center position in x/y at 0 deg. This shift is rotated - using the geometry of LamNI - It is determined by the first 'click' in the x-ray eye alignemnt procedure - angle [deg]: rotation angle (will rotate first) - scan_type: fly (i.e. HW triggered step in case of LamNI) or step - stitch_x/y: shift scan to adjacent stitch region - fov_circular [um]: generate a circular field of view in the sample plane. This is an additional cropping to fov_size. - stitch_overlap [um]: overlap of the stitched regions - Returns: - - Examples: - >>> scans.lamni_fermat_scan(fov_size=[20], step=0.5, exp_time=0.1) - >>> scans.lamni_fermat_scan(fov_size=[20, 25], center_x=0.02, center_y=0, shift_x=0, shift_y=0, angle=0, step=0.5, fov_circular=0, exp_time=0.1, frames_per_trigger=1) - """ - - super().__init__(parameter=parameter, frames_per_trigger=frames_per_trigger, exp_time=exp_time,**kwargs) - self.axis = [] - scan_kwargs = parameter.get("kwargs", {}) - self.fov_size = scan_kwargs.get("fov_size") - if len(self.fov_size) == 1: - self.fov_size *= 2 # if we only have one argument, let's assume it's a square - self.step = scan_kwargs.get("step", 0.1) - self.center_x = scan_kwargs.get("center_x", 0) - self.center_y = scan_kwargs.get("center_y", 0) - self.shift_x = scan_kwargs.get("shift_x", 0) - self.shift_y = scan_kwargs.get("shift_y", 0) - self.angle = scan_kwargs.get("angle", 0) - self.scan_type = scan_kwargs.get("scan_type", "fly") - self.stitch_x = scan_kwargs.get("stitch_x", 0) - self.stitch_y = scan_kwargs.get("stitch_y", 0) - self.fov_circular = scan_kwargs.get("fov_circular", 0) - self.stitch_overlap = scan_kwargs.get("stitch_overlap", 1) - # self.keep_plot = scan_kwargs.get("keep_plot", 0) - self.optim_trajectory = scan_kwargs.get("optim_trajectory", "corridor") - self.optim_trajectory_corridor = scan_kwargs.get("optim_trajectory_corridor") - - def initialize(self): - self.scan_motors = [] - self.update_readout_priority() - - def scan_report_instructions(self): - """Scan report instructions for the progress bar""" - yield from self.stubs.scan_report_instruction({"device_progress": ["rt_positions"]}) - - @property - def monitor_sync(self) -> str: - return "rt_positions" - - def _optimize_trajectory(self): - self.positions = self.optimize_corridor( - self.positions, corridor_size=self.optim_trajectory_corridor - ) - - def prepare_positions(self): - self._calculate_positions() - self._optimize_trajectory() - # self._sort_positions() - - self.num_pos = len(self.positions) - self._check_min_positions() - - def _check_min_positions(self): - if self.num_pos < self._MIN_POSITIONS: - raise ScanAbortion( - f"The number of positions must exceed {self._MIN_POSITIONS}. Currently:" - f" {self.num_pos}." - ) - - @staticmethod - def _lamni_check_pos_in_fov_range_and_circ_fov( - x, y, stitch_x, stitch_y, angle, fov_size, stitch_overlap, fov_circular - ) -> bool: - # this function checks if positions are reachable in a scan - # these x y intererometer positions are not shifted to the scan center - # so its purpose is to see if the position is reachable by the - # rotated piezo stage. For a scan these positions have to be shifted to - # the current scan center before starting the scan - stage_x, stage_y = lamni_to_stage_coordinates(x, y) - stage_x_with_stitch, stage_y_with_stitch = LamNIFermatScan._lamni_compute_stitch_center( - stitch_x, stitch_y, angle, fov_size, stitch_overlap - ) - stage_x_with_stitch, stage_y_with_stitch = lamni_to_stage_coordinates( - stage_x_with_stitch, stage_y_with_stitch - ) - - # piezo stage is currently rotated to stage_angle_deg in degrees - # rotate positions to the piezo stage system - alpha = (angle - 300 + 30.5) / 180 * np.pi - stage_x_rot = np.cos(alpha) * stage_x + np.sin(alpha) * stage_y - stage_y_rot = -np.sin(alpha) * stage_x + np.cos(alpha) * stage_y - - stage_x_rot_with_stitch = ( - np.cos(alpha) * stage_x_with_stitch + np.sin(alpha) * stage_y_with_stitch - ) - stage_y_rot_with_stitch = ( - -np.sin(alpha) * stage_x_with_stitch + np.cos(alpha) * stage_y_with_stitch - ) - - return ( - np.abs(stage_x_rot) <= (fov_size[1] / 2) - and np.abs(stage_y_rot) <= (fov_size[0] / 2) - and ( - fov_circular == 0 - or ( - np.power((stage_x_rot_with_stitch + stage_x_rot), 2) - + np.power((stage_y_rot_with_stitch + stage_y_rot), 2) - ) - <= pow((fov_circular / 2), 2) - ) - ) - - def _prepare_setup(self): - yield from self.stubs.send_rpc_and_wait("rtx", "controller.clear_trajectory_generator") - yield from self.lamni_rotation(self.angle) - total_shift_x, total_shift_y = self._compute_total_shift( - self.center_x, - self.center_y, - self.angle, - self.stitch_x, - self.stitch_y, - self.stitch_overlap, - self.shift_x, - self.shift_y, - self.fov_size, - ) - yield from self.lamni_new_scan_center_interferometer(total_shift_x, total_shift_y) - # self._plot_target_pos() - if self.scan_type == "fly": - yield from self._transfer_positions_to_LamNI() - - # def _plot_target_pos(self): - # # return - # plt.plot(self.positions[:, 0], self.positions[:, 1], alpha=0.2) - # plt.scatter(self.positions[:, 0], self.positions[:, 1]) - # plt.savefig("mygraph.png") - # if not self.keep_plot: - # plt.clf() - # # plt.show() - - def _transfer_positions_to_LamNI(self): - yield from self.stubs.send_rpc_and_wait( - "rtx", "controller.add_pos_to_scan", self.positions.tolist() - ) - - def _calculate_positions(self): - self.positions = self.get_lamni_fermat_spiral_pos( - -np.abs(self.fov_size[0] / 2), - np.abs(self.fov_size[0] / 2), - -np.abs(self.fov_size[1] / 2), - np.abs(self.fov_size[1] / 2), - step=self.step, - spiral_type=0, - center=False, - center_x=self.center_x, - center_y=self.center_y, - angle=self.angle, - stitch_x=self.stitch_x, - stitch_y=self.stitch_y, - stitch_overlap=self.stitch_overlap, - shift_x=self.shift_x, - shift_y=self.shift_y, - fov_size=self.fov_size, - fov_circular=self.fov_circular, - ) - - @staticmethod - def _lamni_compute_stitch_center(xcount, ycount, angle_deg, fov_size, stitch_overlap): - alpha = angle_deg / 180 * np.pi - stage_x = xcount * (fov_size[0] - stitch_overlap) - stage_y = ycount * (fov_size[1] - stitch_overlap) - x_rot = np.cos(alpha) * stage_x - np.sin(alpha) * stage_y - y_rot = np.sin(alpha) * stage_x + np.cos(alpha) * stage_y - - return lamni_from_stage_coordinates(x_rot, y_rot) - - @staticmethod - def _compute_total_shift( - center_x, center_y, angle, stitch_x, stitch_y, stitch_overlap, shift_x, shift_y, fov_size - ): - _shfitx, _shfity = LamNIFermatScan._lamni_compute_scan_center(center_x, center_y, angle) - x_stitch_shift, y_stitch_shift = LamNIFermatScan._lamni_compute_stitch_center( - stitch_x, stitch_y, angle, fov_size, stitch_overlap - ) - logger.info( - f"Total shift [mm] {_shfitx+x_stitch_shift/1000+shift_x}," - f" {_shfity+y_stitch_shift/1000+shift_y}" - ) - return (_shfitx + x_stitch_shift / 1000 + shift_x, _shfity + y_stitch_shift / 1000 + shift_y) - - @staticmethod - def get_lamni_fermat_spiral_pos( - m1_start, - m1_stop, - m2_start, - m2_stop, - step=1, - spiral_type=0, - center=False, - center_x=0.0, - center_y=0.0, - angle=0.0, - stitch_x=0, - stitch_y=0, - stitch_overlap=1, - shift_x=0.0, - shift_y=0.0, - fov_size=None, - fov_circular=0, - ): - """[summary] - - Pure function (no device I/O) -- a @staticmethod rather than an - instance method (center_x/center_y/angle/stitch_x/stitch_y/ - stitch_overlap/shift_x/shift_y/fov_size/fov_circular used to be read - off self.*) so it can also be called directly from client-side code - (tomo_parameters(), tomo_params.py) to predict the point count of a - not-yet-run scan, using the exact same algorithm the scan itself - will use -- including the rotated-stage and circular-FOV cropping - in _lamni_check_pos_in_fov_range_and_circ_fov(). - - Args: - m1_start (float): start position motor 1 - m1_stop (float): end position motor 1 - m2_start (float): start position motor 2 - m2_stop (float): end position motor 2 - step (float, optional): Step size. Defaults to 1. - spiral_type (float, optional): Angular offset in radians that determines the shape of the spiral. - A spiral with spiral_type=2 is the same as spiral_type=0. Defaults to 0. - center (bool, optional): Add a center point. Defaults to False. - fov_size (list): [fov_x, fov_y] used for the rotated-stage/circular-FOV crop. - - Raises: - TypeError: [description] - TypeError: [description] - TypeError: [description] - - Returns: - [type]: [description] - - Yields: - [type]: [description] - """ - positions = [] - phi = 2 * np.pi * ((1 + np.sqrt(5)) / 2.0) + spiral_type * np.pi - - start = int(not center) - - length_axis1 = np.abs(m1_stop - m1_start) - length_axis2 = np.abs(m2_stop - m2_start) - n_max = int(length_axis1 * length_axis2 * 3.2 / step / step) - - total_shift_x, total_shift_y = LamNIFermatScan._compute_total_shift( - center_x, center_y, angle, stitch_x, stitch_y, stitch_overlap, shift_x, shift_y, fov_size - ) - - for ii in range(start, n_max): - radius = step * 0.57 * np.sqrt(ii) - # FOV is restructed below at check pos in range - # if abs(radius * np.sin(ii * phi)) > length_axis1 / 2: - # continue - # if abs(radius * np.cos(ii * phi)) > length_axis2 / 2: - # continue - x = radius * np.sin(ii * phi) - y = radius * np.cos(ii * phi) - if LamNIFermatScan._lamni_check_pos_in_fov_range_and_circ_fov( - x, y, stitch_x, stitch_y, angle, fov_size, stitch_overlap, fov_circular - ): - positions.extend([(x + total_shift_x * 1000, y + total_shift_y * 1000)]) - # for testing we just shift by center_i and prepare also the setup to center_i - return np.array(positions) - - def lamni_rotation(self, angle): - # get last setpoint (cannot be based on pos get because they will deviate slightly) - lsamrot_current_setpoint = yield from self.stubs.send_rpc_and_wait( - "lsamrot", "user_setpoint.get" - ) - if angle == lsamrot_current_setpoint: - logger.info("No rotation required") - else: - logger.info("Rotating to requested angle") - yield from self.stubs.scan_report_instruction( - { - "readback": { - "RID": self.metadata["RID"], - "devices": ["lsamrot"], - "start": [lsamrot_current_setpoint], - "end": [angle], - } - } - ) - yield from self.stubs.set(device="lsamrot", value=angle) - - def scan_core(self): - # fly scan mode - yield from self.stubs.kickoff(device="rt_positions") - - # start the readout loop of the flyer - status = yield from self.stubs.complete(device="rt_positions", wait=False) - - while not status.done: - yield from self.stubs.read(group="monitored", point_id=self.point_id) - self.point_id += 1 - time.sleep(1) - logger.debug("reading monitors") - - - def run(self): - self.initialize() - yield from self.read_scan_motors() - self.prepare_positions() - yield from self._prepare_setup() - yield from self.scan_report_instructions() - yield from self.open_scan() - yield from self.stage() - yield from self.run_baseline_reading() - yield from self.pre_scan() - yield from self.scan_core() - yield from self.finalize() - yield from self.unstage() - yield from self.cleanup() diff --git a/csaxs_bec/scans/__init__.py b/csaxs_bec/scans/__init__.py index ba6d225..4d01624 100644 --- a/csaxs_bec/scans/__init__.py +++ b/csaxs_bec/scans/__init__.py @@ -1,7 +1,8 @@ from .flomni_fermat_scan import FlomniFermatScan from .jungfrau_joch_scan import JungfrauJochTestScan -from .LamNIFermatScan import LamNIFermatScan, LamNIMoveToScanCenter -from .omny_fermat_scan import OMNYFermatScan +from .lamni_fermat_scan import LamniFermatScan +from .lamni_move_to_scan_center import LamniMoveToScanCenter +from .omny_fermat_scan import OmnyFermatScan from .owis_grid import OwisGrid from .scans_v4.cont_grid import ContGrid from .sgalil_grid import SgalilGrid diff --git a/csaxs_bec/scans/flomni_fermat_scan.py b/csaxs_bec/scans/flomni_fermat_scan.py index f3c6d41..c063dd4 100644 --- a/csaxs_bec/scans/flomni_fermat_scan.py +++ b/csaxs_bec/scans/flomni_fermat_scan.py @@ -1,94 +1,106 @@ """ -SCAN PLUGINS +flOMNI Fermat Scan -All new scans should be derived from ScanBase. ScanBase provides various methods that can be customized and overriden -but they are executed in a specific order: - -- self.initialize # initialize the class if needed -- self.read_scan_motors # used to retrieve the start position (and the relative position shift if needed) -- self.prepare_positions # prepare the positions for the scan. The preparation is split into multiple sub fuctions: - - self._calculate_positions # calculate the positions - - self._set_positions_offset # apply the previously retrieved scan position shift (if needed) - - self._check_limits # tests to ensure the limits won't be reached -- self.open_scan # send an open_scan message including the scan name, the number of points and the scan motor names -- self.stage # stage all devices for the upcoming acquisiton -- self.run_baseline_readings # read all devices to get a baseline for the upcoming scan -- self.scan_core # run a loop over all position - - self._at_each_point(ind, pos) # called at each position with the current index and the target positions as arguments -- self.finalize # clean up the scan, e.g. move back to the start position; wait everything to finish -- self.unstage # unstage all devices that have been staged before -- self.cleanup # send a close scan message and perform additional cleanups if needed +Scan procedure: + - prepare_scan + - open_scan + - stage + - pre_scan + - scan_core + - at_each_point (optionally called by scan_core) + - post_scan + - unstage + - close_scan + - on_exception (called if any exception is raised during the scan) """ +from __future__ import annotations + import time +from typing import Annotated import numpy as np -from bec_lib import bec_logger, messages +from bec_lib import messages from bec_lib.alarm_handler import Alarms from bec_lib.endpoints import MessageEndpoints -from bec_server.scan_server.errors import ScanAbortion -from bec_server.scan_server.scans import AsyncFlyScanBase +from bec_lib.logger import bec_logger +from bec_lib.scan_args import DefaultArgType, ScanArgument, Units +from bec_server.scan_server.scans import ScanAbortion +from bec_server.scan_server.scans.scan_base import ScanBase, ScanType +from bec_server.scan_server.scans.scan_modifier import scan_hook from csaxs_bec.devices.epics.delay_generator_csaxs.delay_generator_csaxs import TRIGGERSOURCE logger = bec_logger.logger -class FlomniFermatScan(AsyncFlyScanBase): - scan_name = "flomni_fermat_scan" - scan_type = "fly" - required_kwargs = ["fovx", "fovy", "exp_time", "step", "angle"] - arg_input = {} - arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None} +class FlomniFermatScan(ScanBase): + # Scan Type: Hardware triggered or software triggered? + # If the main trigger and readout logic is done within the at_each_point method in scan_core, choose SOFTWARE_TRIGGERED. + # If the main trigger and readout logic is implemented on a device that is simply kicked off in this scan, choose HARDWARE_TRIGGERED. + # This primarily serves as information for devices: The device may need to react differently if a software trigger is expected + # for every point. + scan_type = ScanType.HARDWARE_TRIGGERED - # Minimum number of Fermat-spiral positions a scan is allowed to run - # with -- exposed as a class attribute (not just a literal inside - # _check_min_positions()) so client-side code (tomo_parameters(), - # tomo_params.py) can warn about a too-few-points configuration before - # it ever reaches the scan server, using the exact same threshold. - _MIN_POSITIONS = 20 + # Scan name: This is the name of the scan, e.g. "line_scan". This is used for display purposes and to identify the scan type in user interfaces. + # Choose a descriptive name that does not conflict with existing scan names. + # It must be a valid Python identifier, that is, it can only contain letters, numbers, and underscores, and must not start with a number. + scan_name = "flomni_fermat_scan" + + gui_config = { + "Scan Parameters": [ + "fovx", + "fovy", + "cenx", + "ceny", + "step", + "zshift", + "angle", + "corridor_size", + ], + "Acquisition Parameters": ["exp_time", "frames_per_trigger", "burst_at_each_point"], + } + + MIN_POSITIONS = 20 # Minimum number of positions required for a valid scan def __init__( + # fmt: off self, - fovx: float, - fovy: float, - cenx: float, - ceny: float, - exp_time: float, - step: float, - zshift: float, - angle: float = None, - corridor_size: float = 3, - parameter: dict = None, - frames_per_trigger: int = 1, + fovx: Annotated[float, ScanArgument(display_name="Fovx", description="FOV in the piezo plane (i.e. piezo range). Max 200 um.", units=Units.µm, gt=0, lt=200)], + fovy: Annotated[float, ScanArgument(display_name="Fovy", description="FOV in the piezo plane (i.e. piezo range). Max 100 um.", units=Units.µm, gt=0, lt=100)], + cenx: Annotated[float, ScanArgument(display_name="Cenx", description="Center position in x.", units=Units.µm)], + ceny: Annotated[float, ScanArgument(display_name="Ceny", description="Center position in y.", units=Units.µm)], + step: Annotated[float, ScanArgument(display_name="Step", description="Step size.", units=Units.µm)], + zshift: Annotated[float, ScanArgument(display_name="Zshift", description="Shift in z. ", units=Units.µm)], + angle: Annotated[float, ScanArgument(display_name="Angle", description="Rotation angle (will rotate first)", units=Units.deg)], + corridor_size: Annotated[float | None, ScanArgument(display_name="Corridor Size", description="Corridor size for the corridor optimization.", units=Units.µm)] = None, + exp_time: DefaultArgType.ExposureTime = 0, + frames_per_trigger: DefaultArgType.FramesPerTrigger = 1, + burst_at_each_point: DefaultArgType.BurstAtEachPoint = 1, **kwargs, + # fmt: on ): """ - A flomni scan following Fermat's spiral. + flOMNI Fermat Scan Args: - fovx(float) [um]: Fov in the piezo plane (i.e. piezo range). Max 200 um - fovy(float) [um]: Fov in the piezo plane (i.e. piezo range). Max 100 um - cenx(float) [um]: center position in x. - ceny(float) [um]: center position in y. - exp_time(float) [s]: exposure time per burst frame - frames_per_trigger(int) : Number of burst frames per point - step(float) [um]: stepsize - zshift(float) [um]: shift in z - angle(float) [deg]: rotation angle (will rotate first) - corridor_size(float) [um]: corridor size for the corridor optimization. Default 3 um + fovx (float): FOV in the piezo plane (i.e. piezo range). Max 200 um. + fovy (float): FOV in the piezo plane (i.e. piezo range). Max 100 um. + cenx (float): Center position in x. + ceny (float): Center position in y. + step (float): Step size. + zshift (float): Shift in z. + angle (float): Rotation angle (will rotate first) + corridor_size (float | None): Corridor size for the corridor optimization. + exp_time (float): Exposure time in seconds + frames_per_trigger (int): Number of frames per trigger for devices that support configurable frame counts per trigger. + burst_at_each_point (int): Number of triggers and readouts at each point. Returns: - - Examples: - >>> scans.flomni_fermat_scan(fovx=20, fovy=25, cenx=0.02, ceny=0, zshift=0, angle=0, step=0.5, exp_time=0.01, frames_per_trigger=1) + ScanReport """ - - super().__init__( - parameter=parameter, exp_time=exp_time, frames_per_trigger=frames_per_trigger, **kwargs - ) - self.show_live_table = False - self.axis = [] + super().__init__(**kwargs) + self._baseline_readout_status = None self.fovx = fovx self.fovy = fovy self.cenx = cenx @@ -96,105 +108,229 @@ class FlomniFermatScan(AsyncFlyScanBase): self.step = step self.zshift = zshift self.angle = angle - self.optim_trajectory = "corridor" - self.optim_trajectory_corridor = corridor_size - if self.fovy > 100: - raise ScanAbortion("The FOV in y must be smaller than 100 um.") - if self.fovx > 200: - raise ScanAbortion("The FOV in x must be smaller than 200 um.") + self.corridor_size = corridor_size + self.exp_time = exp_time + self.frames_per_trigger = frames_per_trigger + self.burst_at_each_point = burst_at_each_point + self.flomni_rotation_status = None + if self.zshift > 100: logger.warning("The zshift is larger than 100 um. It will be limited to 100 um.") self.zshift = 100 + if self.zshift < -100: logger.warning("The zshift is smaller than -100 um. It will be limited to -100 um.") self.zshift = -100 - self.flomni_rotation_status = None - def scan_report_instructions(self): - """Scan report instructions for the progress bar""" - yield from self.stubs.scan_report_instruction({"device_progress": ["rt_positions"]}) - - @property - def monitor_sync(self) -> str: - return "rt_positions" - - def initialize(self): - self.scan_motors = [] - self.update_readout_priority() - - def _optimize_trajectory(self): - self.positions = self.optimize_corridor( - self.positions, corridor_size=self.optim_trajectory_corridor + self.update_scan_info( + exp_time=exp_time, + frames_per_trigger=frames_per_trigger, + burst_at_each_point=burst_at_each_point, ) + @scan_hook + def prepare_scan(self): + """ + Prepare the scan. This can include any steps that need to be executed + before the scan is opened, such as preparing the positions (if not done already) + or setting up the devices. + """ + + positions = self.get_flomni_fermat_spiral_pos( + -np.abs(self.fovx / 2), + np.abs(self.fovx / 2), + -np.abs(self.fovy / 2), + np.abs(self.fovy / 2), + step=self.step, + spiral_type=0, + center=False, + cenx=self.cenx, + ceny=self.ceny, + zshift=self.zshift, + ) + + if len(positions) < self.MIN_POSITIONS: + raise ScanAbortion( + f"The number positions must exceed {self.MIN_POSITIONS}. Currently: {len(positions)}." + ) + + self.positions = self.components.optimize_trajectory( + positions, optimization_type="corridor", corridor_size=self.corridor_size + ) + flip_axes = self.reverse_trajectory() + if flip_axes: + self.positions = np.flipud(self.positions) + + self.update_scan_info(positions=self.positions, num_points=len(self.positions)) + + self.prepare_setup() + + self.actions.add_scan_report_instruction_device_progress(device=self.dev.rt_positions) + self._baseline_readout_status = self.actions.read_baseline_devices(wait=False) + + @scan_hook + def open_scan(self): + """ + Open the scan. + This step must call self.actions.open_scan() to ensure that a new scan is + opened. Make sure to prepare the scan metadata before, either in + prepare_scan() or in open_scan() itself and call self.update_scan_info(...) + to update the scan metadata if needed. + """ + self.actions.open_scan() + + @scan_hook + def stage(self): + """ + Stage the devices for the upcoming scan. The stage logic is typically + implemented on the device itself (i.e. by the device's stage method). + However, if there are any additional steps that need to be executed before + staging the devices, they can be implemented here. + """ + self.actions.stage_all_devices() + + @scan_hook + def pre_scan(self): + """ + Pre-scan steps to be executed before the main scan logic. + This is typically the last chance to prepare the devices before the core scan + logic is executed. For example, this is a good place to initialize time-criticial + devices, e.g. devices that have a short timeout. + The pre-scan logic is typically implemented on the device itself. + """ + self.prepare_setup_part_2() + self.actions.pre_scan_all_devices() + + @scan_hook + def scan_core(self): + """ + Core scan logic to be executed during the scan. + This is where the main scan logic should be implemented. + """ + + # send off the flyer + self.actions.kickoff(device=self.dev.rt_positions) + + # start the readout loop of the flyer + status = self.actions.complete(device=self.dev.rt_positions, wait=False) + while not status.done: + self.at_each_point() + + @scan_hook + def at_each_point(self): + """ + Logic to be executed at each acquisition point during the scan. + """ + self.actions.read_monitored_devices() + time.sleep(1) + + @scan_hook + def post_scan(self): + """ + Post-scan steps to be executed after the main scan logic. + """ + # in flomni, we need to move to the start position of the next scan, + # which is the end position of the current scan + move_status = None + if isinstance(self.positions, np.ndarray) and len(self.positions[-1]) == 3: + # in x we move to cenx, then we avoid jumps in centering routine + value = self.positions[-1] + value[0] = self.cenx + move_status = self.actions.set(device=["rtx", "rty", "rtz"], value=value, wait=False) + + self.actions.complete_all_devices() + + if move_status: + move_status.wait() + + self.dev.ddg1.set_trigger(TRIGGERSOURCE.SINGLE_SHOT.value) + + @scan_hook + def unstage(self): + """Unstage the scan by executing post-scan steps.""" + self.actions.unstage_all_devices() + + @scan_hook + def close_scan(self): + """Close the scan.""" + if self._baseline_readout_status is not None: + self._baseline_readout_status.wait() + self.actions.close_scan() + self.actions.check_for_unchecked_statuses() + + @scan_hook + def on_exception(self, exception: Exception): + """ + Handle exceptions that occur during the scan. + This is a good place to implement any cleanup logic that needs to be executed in case of an exception, + such as returning the devices to a safe state or moving the motors back to their starting position. + """ + self.dev.ddg1.set_trigger(TRIGGERSOURCE.SINGLE_SHOT.value) + + ####################################################### + ######### Helper methods for the scan logic ########### + ####################################################### + def reverse_trajectory(self): """ Reverse the trajectory. Every other scan should be reversed to shorten the movement time. In order to keep the last state, even if the server is restarted, the state is stored in a global variable in redis. """ - msg = self.connector.get(MessageEndpoints.global_vars("reverse_flomni_trajectory")) + msg = self.redis_connector.get(MessageEndpoints.global_vars("reverse_flomni_trajectory")) if msg: val = msg.content.get("value", False) else: val = False - self.connector.set( + self.redis_connector.set( MessageEndpoints.global_vars("reverse_flomni_trajectory"), messages.VariableMessage(value=(not val)), ) return val - def prepare_positions(self): - self._calculate_positions() - self._optimize_trajectory() - flip_axes = self.reverse_trajectory() - if flip_axes: - self.positions = np.flipud(self.positions) + def prepare_setup(self): + """ + Prepare the first part of the setup: + - Clear the trajectory of the rt controller + - Rotate flomni to the requested angle + - Move rty to the start position + """ + self.dev.rtx.controller.clear_trajectory_generator() + self.flomni_rotation(self.angle) + self.actions.set(self.dev.rty, self.positions[0][1]) - self.num_pos = len(self.positions) - self._check_min_positions() + def prepare_setup_part_2(self): + """ + Prepare the second part of the setup: + - Set the delay generator ddg1 to external rising edge + - Wait for flomni rotation to complete (started in prepare_setup) + - Move rtx and rtz to the start position + - Turn on the laser tracker + - Add the positions to the rt controller's scan trajectory + - Check the signal strength of the laser tracker and raise an alarm if it is low + - Move samx to the scan region + """ + dev = self.dev - def _check_min_positions(self): - if self.num_pos < self._MIN_POSITIONS: - raise ScanAbortion( - f"The number of positions must exceed {self._MIN_POSITIONS}. Currently:" - f" {self.num_pos}." - ) - - def _prepare_setup(self): - yield from self.stubs.send_rpc_and_wait("rtx", "controller.clear_trajectory_generator") - yield from self.flomni_rotation(self.angle) - - yield from self.stubs.send_rpc_and_wait("rty", "set", self.positions[0][1]) - - def _prepare_setup_part2(self): - # Prepare DDG1 to use - yield from self.stubs.send_rpc_and_wait( - "ddg1", "set_trigger", TRIGGERSOURCE.EXT_RISING_EDGE.value - ) + # Prepare DDG1 + dev.ddg1.set_trigger(TRIGGERSOURCE.EXT_RISING_EDGE.value) if self.flomni_rotation_status: self.flomni_rotation_status.wait() - # rtx_status = yield from self.stubs.set(device="rtx", value=self.positions[0][0], wait=False) - rtx_status = yield from self.stubs.set(device="rtx", value=self.cenx, wait=False) - rtz_status = yield from self.stubs.set(device="rtz", value=self.positions[0][2], wait=False) + rtx_status = dev.rtx.set(self.cenx) + rtz_status = dev.rtz.set(self.positions[0][2]) - yield from self.stubs.send_rpc_and_wait("rtx", "controller.laser_tracker_on") + dev.rtx.controller.laser_tracker_on() rtx_status.wait() rtz_status.wait() - # status = yield from self.stubs.send_rpc("rtx", "move", self.cenx) - # status.wait() - yield from self._transfer_positions_to_flomni() - tracker_signal_status = yield from self.stubs.send_rpc_and_wait( - "rtx", "controller.laser_tracker_check_signalstrength" - ) - yield from self.stubs.send_rpc_and_wait( - "rtx", "controller.move_samx_to_scan_region", self.cenx - ) - # self.device_manager.connector.send_client_info(tracker_signal_status) + dev.rtx.controller.add_pos_to_scan(self.positions.tolist()) + + tracker_signal_status = dev.rtx.controller.laser_tracker_check_signalstrength() + dev.rtx.controller.move_samx_to_scan_region(self.cenx) + if tracker_signal_status == "low": error_info = messages.ErrorInfo( error_message="Signal strength of the laser tracker is low, but sufficient to continue. Realignment recommended!", @@ -208,70 +344,48 @@ class FlomniFermatScan(AsyncFlyScanBase): "Signal strength of the laser tracker is too low for scanning. Realignment required!" ) - def flomni_rotation(self, angle): - # get last setpoint (cannot be based on pos get because they will deviate slightly) - fsamroy_current_setpoint = yield from self.stubs.send_rpc_and_wait( - "fsamroy", "user_setpoint.get" - ) + def flomni_rotation(self, angle: float): + """ + Rotate flomni to the requested angle. + We also emit a scan report instruction to keep users informed about the progress of the + rotation as it may take a few seconds. + + Note that we do not wait for the rotation to complete here, but + instead wait in prepare_setup_part_2. + + Args: + angle (float): The target angle for the flomni rotation. + """ + fsamroy_current_setpoint = self.dev.fsamroy.user_setpoint.get() if angle == fsamroy_current_setpoint: logger.info("No rotation required") - else: - logger.info("Rotating to requested angle") - yield from self.stubs.scan_report_instruction( - { - "readback": { - "RID": self.metadata["RID"], - "devices": ["fsamroy"], - "start": [fsamroy_current_setpoint], - "end": [angle], - } - } - ) - self.flomni_rotation_status = yield from self.stubs.set( - device="fsamroy", value=angle, wait=False - ) + return - def _transfer_positions_to_flomni(self): - yield from self.stubs.send_rpc_and_wait( - "rtx", "controller.add_pos_to_scan", self.positions.tolist() - ) - - def _calculate_positions(self): - self.positions = self.get_flomni_fermat_spiral_pos( - -np.abs(self.fovx / 2), - np.abs(self.fovx / 2), - -np.abs(self.fovy / 2), - np.abs(self.fovy / 2), - step=self.step, - spiral_type=0, - center=False, - cenx=self.cenx, - ceny=self.ceny, - zshift=self.zshift, + logger.info("Rotating to requested angle") + self.actions.add_scan_report_instruction_readback( + devices=["fsamroy"], + start=[fsamroy_current_setpoint], + stop=[angle], + request_id=self.scan_info.metadata["RID"], ) + self.flomni_rotation_status = self.actions.set(self.dev.fsamroy, angle, wait=False) @staticmethod def get_flomni_fermat_spiral_pos( - m1_start, - m1_stop, - m2_start, - m2_stop, - step=1, - spiral_type=0, - center=False, - cenx=0.0, - ceny=0.0, - zshift=0.0, + m1_start: float, + m1_stop: float, + m2_start: float, + m2_stop: float, + step: float = 1, + spiral_type: int = 0, + center: bool = False, + cenx: float = 0.0, + ceny: float = 0.0, + zshift: float = 0.0, ): """ Calculate positions for a Fermat spiral scan. - Pure function (no device I/O) -- a @staticmethod rather than an - instance method (cenx/ceny/zshift used to be read off self.*) so it - can also be called directly from client-side code (tomo_parameters(), - tomo_params.py) to predict the point count of a not-yet-run scan, - using the exact same algorithm the scan itself will use. - Args: m1_start(float): start position in m1 m1_stop(float): stop position in m1 @@ -280,9 +394,6 @@ class FlomniFermatScan(AsyncFlyScanBase): step(float): stepsize spiral_type(int): 0 for traditional Fermat spiral center(bool): whether to include the center position - cenx(float): center offset added to every x position - ceny(float): center offset added to every y position - zshift(float): z position for every point Returns: positions(array): positions @@ -313,53 +424,3 @@ class FlomniFermatScan(AsyncFlyScanBase): positions.append(left_lower_corner) positions.append(right_upper_corner) return np.array(positions) - - def scan_core(self): - # send off the flyer - yield from self.stubs.kickoff(device="rt_positions") - - # start the readout loop of the flyer - status = yield from self.stubs.complete(device="rt_positions", wait=False) - - # read the monitors until the flyer is done - while not status.done: - yield from self.stubs.read(group="monitored", point_id=self.point_id) - self.point_id += 1 - time.sleep(1) - logger.debug("reading monitors") - - def move_to_start(self): - """return to the start position""" - # in flomni, we need to move to the start position of the next scan, which is the end position of the current scan - # this method is called in finalize and overwrites the default move_to_start() - if isinstance(self.positions, np.ndarray) and len(self.positions[-1]) == 3: - # yield from self.stubs.set(device=["rtx", "rty", "rtz"], value=self.positions[-1]) - # in x we move to cenx, then we avoid jumps in centering routine - value = self.positions[-1] - value[0] = self.cenx - yield from self.stubs.set(device=["rtx", "rty", "rtz"], value=value) - return - - logger.warning("No positions found to return to start") - - def cleanup(self): - yield from self.stubs.send_rpc_and_wait( - "ddg1", "set_trigger", TRIGGERSOURCE.SINGLE_SHOT.value - ) - yield from super().cleanup() - - def run(self): - self.initialize() - yield from self.read_scan_motors() - self.prepare_positions() - yield from self._prepare_setup() - yield from self.scan_report_instructions() - yield from self.open_scan() - yield from self.stage() - yield from self.run_baseline_reading() - yield from self._prepare_setup_part2() - yield from self.pre_scan() - yield from self.scan_core() - yield from self.finalize() - yield from self.unstage() - yield from self.cleanup() diff --git a/csaxs_bec/scans/lamni_components.py b/csaxs_bec/scans/lamni_components.py new file mode 100644 index 0000000..3c9648a --- /dev/null +++ b/csaxs_bec/scans/lamni_components.py @@ -0,0 +1,179 @@ +from __future__ import annotations + +import time + +import numpy as np +from bec_lib.logger import bec_logger +from bec_server.scan_server.scans.scan_components import ScanComponents + +logger = bec_logger.logger + +MOVEMENT_SCALE_X = np.sin(np.radians(15)) * np.cos(np.radians(30)) +MOVEMENT_SCALE_Y = np.cos(np.radians(15)) + + +class LamNIComponents(ScanComponents): + + @staticmethod + def lamni_compute_scan_center(x: float, y: float, angle_deg: float) -> tuple[float, float]: + """ + Compute the scan center in the stage coordinates based on the + provided center in lamni coordinates and the rotation angle. + + Args: + x (float): Center position in x at 0 deg in lamni coordinates (mm) + y (float): Center position in y at 0 deg in lamni coordinates (mm) + angle_deg (float): Rotation angle in degrees + + Returns: + tuple: (shift_x, shift_y) in mm to be applied to the scan center in the stage coordinates + """ + alpha = angle_deg / 180 * np.pi + stage_x, stage_y = LamNIComponents.lamni_to_stage_coordinates(x, y) + stage_x_rot = np.cos(alpha) * stage_x - np.sin(alpha) * stage_y + stage_y_rot = np.sin(alpha) * stage_x + np.cos(alpha) * stage_y + return LamNIComponents.lamni_from_stage_coordinates(stage_x_rot, stage_y_rot) + + @staticmethod + def lamni_to_stage_coordinates(x: float, y: float) -> tuple[float, float]: + """convert from lamni coordinates to stage coordinates""" + y_stage = y / MOVEMENT_SCALE_Y + x_stage = 2 * (x - y_stage * MOVEMENT_SCALE_X) + return (x_stage, y_stage) + + @staticmethod + def lamni_from_stage_coordinates(x_stage: float, y_stage: float) -> tuple[float, float]: + """convert to lamni coordinates from stage coordinates""" + x = x_stage * 0.5 + y_stage * MOVEMENT_SCALE_X + y = y_stage * MOVEMENT_SCALE_Y + return (x, y) + + def lamni_new_scan_center_interferometer(self, x: float, y: float): + """ + Move to the new scan center. + + Args: + x (float): Center position in x in mm + y (float): Center position in y in mm + """ + lsamx_user_params = self._dev.lsamx.user_parameter + if lsamx_user_params is None or lsamx_user_params.get("center") is None: + raise RuntimeError("lsamx center is not defined") + lsamy_user_params = self._dev.lsamy.user_parameter + + if lsamy_user_params is None or lsamy_user_params.get("center") is None: + raise RuntimeError("lsamy center is not defined") + + lsamx_center = lsamx_user_params.get("center") + lsamy_center = lsamy_user_params.get("center") + + # disable the feedback + self._dev.rtx.controller.feedback_disable() + + rtx_current = self._dev.rtx.readback.get() + rty_current = self._dev.rty.readback.get() + lsamx_current = self._dev.lsamx.readback.get() + lsamy_current = self._dev.lsamy.readback.get() + + x_stage, y_stage = self.lamni_to_stage_coordinates(x, y) + x_center_expect, y_center_expect = self.lamni_from_stage_coordinates( + lsamx_current - lsamx_center, lsamy_current - lsamy_center + ) + + # in microns + x_drift = x_center_expect * 1000 - rtx_current + y_drift = y_center_expect * 1000 - rty_current + + logger.info(f"Current uncompensated drift of setup is x={x_drift:.3f}, y={y_drift:.3f}") + + move_x = ( + x_stage + lsamx_center + self.lamni_to_stage_coordinates(x_drift, y_drift)[0] / 1000 + ) + move_y = ( + y_stage + lsamy_center + self.lamni_to_stage_coordinates(x_drift, y_drift)[1] / 1000 + ) + + coarse_move_req_x = np.abs(lsamx_current - move_x) + coarse_move_req_y = np.abs(lsamy_current - move_y) + + self._dev.lsamx.read_only = False + self._dev.lsamy.read_only = False + + if ( + np.abs(y_drift) > 150 + or np.abs(x_drift) > 150 + or (coarse_move_req_y < 0.003 and coarse_move_req_x < 0.003) + ): + logger.info("No drift correction.") + else: + logger.info( + f"Compensating {[val/1000 for val in self.lamni_to_stage_coordinates(x_drift,y_drift)]}" + ) + self._dev.lsamx.set(move_x).wait() + self._dev.lsamy.set(move_y).wait() + + time.sleep(0.01) + rtx_current = self._dev.rtx.readback.get() + rty_current = self._dev.rty.readback.get() + + logger.info(f"New scan center interferometer {rtx_current:.3f}, {rty_current:.3f} microns") + + # second iteration + x_center_expect, y_center_expect = self.lamni_from_stage_coordinates(x_stage, y_stage) + + # in microns + x_drift2 = x_center_expect * 1000 - rtx_current + y_drift2 = y_center_expect * 1000 - rty_current + logger.info( + f"Uncompensated drift of setup after first iteration is x={x_drift2:.3f}," + f" y={y_drift2:.3f}" + ) + + if np.abs(x_drift2) > 5 or np.abs(y_drift2) > 5: + logger.info( + "Compensating second iteration" + f" {[val/1000 for val in self.lamni_to_stage_coordinates(x_drift2,y_drift2)]}" + ) + move_x = ( + x_stage + + lsamx_center + + self.lamni_to_stage_coordinates(x_drift, y_drift)[0] / 1000 + + self.lamni_to_stage_coordinates(x_drift2, y_drift2)[0] / 1000 + ) + move_y = ( + y_stage + + lsamy_center + + self.lamni_to_stage_coordinates(x_drift, y_drift)[1] / 1000 + + self.lamni_to_stage_coordinates(x_drift2, y_drift2)[1] / 1000 + ) + lsamx_set = self._dev.lsamx.set(move_x) + lsamy_set = self._dev.lsamy.set(move_y) + + lsamx_set.wait() + lsamy_set.wait() + + time.sleep(0.01) + rtx_current = self._dev.rtx.readback.get() + rty_current = self._dev.rty.readback.get() + + logger.info( + f"New scan center interferometer after second iteration {rtx_current:.3f}," + f" {rty_current:.3f} microns" + ) + x_drift2 = x_center_expect * 1000 - rtx_current + y_drift2 = y_center_expect * 1000 - rty_current + logger.info( + f"Uncompensated drift of setup after second iteration is x={x_drift2:.3f}," + f" y={y_drift2:.3f}" + ) + else: + logger.info("No second iteration required") + + self._dev.lsamx.read_only = True + self._dev.lsamy.read_only = True + + # update angle readback before start of the scan + self._dev.lsamrot.readback.get() + + # re-enable the feedback + self._dev.rtx.controller.feedback_enable_without_reset() diff --git a/csaxs_bec/scans/lamni_fermat_scan.py b/csaxs_bec/scans/lamni_fermat_scan.py new file mode 100644 index 0000000..7ef4394 --- /dev/null +++ b/csaxs_bec/scans/lamni_fermat_scan.py @@ -0,0 +1,525 @@ +""" +LamNI Fermat Scan + +Scan procedure: + - prepare_scan + - open_scan + - stage + - pre_scan + - scan_core + - at_each_point (optionally called by scan_core) + - post_scan + - unstage + - close_scan + - on_exception (called if any exception is raised during the scan) +""" + +from __future__ import annotations + +import time +from typing import Annotated + +import numpy as np +from bec_lib.logger import bec_logger +from bec_lib.scan_args import DefaultArgType, ScanArgument, Units +from bec_server.scan_server.errors import ScanAbortion +from bec_server.scan_server.scans.scan_base import ScanBase, ScanType +from bec_server.scan_server.scans.scan_modifier import scan_hook + +from csaxs_bec.scans.lamni_components import LamNIComponents + +logger = bec_logger.logger + + +class LamniFermatScan(ScanBase): + # Scan Type: Hardware triggered or software triggered? + # If the main trigger and readout logic is done within the at_each_point method in scan_core, choose SOFTWARE_TRIGGERED. + # If the main trigger and readout logic is implemented on a device that is simply kicked off in this scan, choose HARDWARE_TRIGGERED. + # This primarily serves as information for devices: The device may need to react differently if a software trigger is expected + # for every point. + scan_type = ScanType.HARDWARE_TRIGGERED + + # Scan name: This is the name of the scan, e.g. "line_scan". This is used for display purposes and to identify the scan type in user interfaces. + # Choose a descriptive name that does not conflict with existing scan names. + # It must be a valid Python identifier, that is, it can only contain letters, numbers, and underscores, and must not start with a number. + scan_name = "lamni_fermat_scan" + + gui_config = { + "Scan Parameters": [ + "fovx", + "fovy", + "step", + "shift_x", + "shift_y", + "center_x", + "center_y", + "angle", + "stitch_x", + "stitch_y", + "fov_circular", + "stitch_overlap", + ], + "Acquisition Parameters": ["exp_time", "frames_per_trigger", "readout_time"], + } + + MIN_POSITIONS = 20 # Minimum number of positions required for a valid scan + + def __init__( + # fmt: off + self, + fovx: Annotated[float, ScanArgument(display_name="FOV x", description="FOV in the piezo plane (i.e. piezo range). Max 80 um", units=Units.µm, gt=0, lt=80)], + fovy: Annotated[float, ScanArgument(display_name="FOV y", description="FOV in the piezo plane (i.e. piezo range). Max 80 um", units=Units.µm, gt=0, lt=80)], + step: Annotated[float, ScanArgument(display_name="Step", description="Step size", units=Units.µm)], + shift_x: Annotated[float, ScanArgument(display_name="Shift X", description="Extra shift in x. The shift is directly applied to the scan. It will not be auto-rotated.", units=Units.mm)] = 0, + shift_y: Annotated[float, ScanArgument(display_name="Shift Y", description="Extra shift in y. The shift is directly applied to the scan. It will not be auto-rotated.", units=Units.mm)] = 0, + center_x: Annotated[float, ScanArgument(display_name="Center X", description="Center position in x at 0 deg. This shift is rotated using the geometry of LamNI. It is determined by the first 'click' in the x-ray eye alignment procedure.", units=Units.mm)] = 0, + center_y: Annotated[float, ScanArgument(display_name="Center Y", description="Center position in y at 0 deg. This shift is rotated using the geometry of LamNI. It is determined by the first 'click' in the x-ray eye alignment procedure.", units=Units.mm)] = 0, + angle: Annotated[float, ScanArgument(display_name="Angle", description="Rotation angle (will rotate first)", units=Units.deg)] = 0, + stitch_x: Annotated[float, ScanArgument(display_name="Stitch X", description="Shift scan to adjacent stitch region", units=Units.mm)] = 0, + stitch_y: Annotated[float, ScanArgument(display_name="Stitch Y", description="Shift scan to adjacent stitch region", units=Units.mm)] = 0, + fov_circular: Annotated[float, ScanArgument(display_name="Fov Circular", description="Generate a circular field of view in the sample plane. This is an additional cropping within the rectangular fovx/fovy range", units=Units.µm)] = 0, + stitch_overlap: Annotated[float, ScanArgument(display_name="Stitch Overlap", description="Overlap of the stitched regions", units=Units.µm)] = 1, + corridor_size: Annotated[float | None, ScanArgument(display_name="Corridor Size", description="Corridor size for the corridor optimization.", units=Units.µm)] = None, + exp_time: DefaultArgType.ExposureTime = 0, + frames_per_trigger: DefaultArgType.FramesPerTrigger = 1, + readout_time: DefaultArgType.ReadoutTime = 0, + **kwargs, + # fmt: on + ): + """ + LamNI Fermat Scan + + Args: + fovx (float): FOV in the piezo plane (i.e. piezo range) along the x-axis. Max 80 um + fovy (float): FOV in the piezo plane (i.e. piezo range) along the y-axis. Max 80 um + step (float): Step size + shift_x (float): Extra shift in x. The shift is directly applied to the scan. It will not be auto-rotated. + shift_y (float): Extra shift in y. The shift is directly applied to the scan. It will not be auto-rotated. + center_x (float): Center position in x at 0 deg. This shift is rotated using the geometry of LamNI. It is determined by the first 'click' in the x-ray eye alignment procedure. + center_y (float): Center position in y at 0 deg. This shift is rotated using the geometry of LamNI. It is determined by the first 'click' in the x-ray eye alignment procedure. + angle (float): Rotation angle (will rotate first) + stitch_x (float): Shift scan to adjacent stitch region + stitch_y (float): Shift scan to adjacent stitch region + fov_circular (float): Generate a circular field of view in the sample plane. This is an additional cropping within the rectangular fovx/fovy range + stitch_overlap (float): Overlap of the stitched regions + exp_time (float): Exposure time in seconds + corridor_size (float | None): Corridor size for the corridor optimization. If None, the corridor size will be estimated. + frames_per_trigger (int): Number of frames per trigger for devices that support configurable frame counts per trigger. + readout_time (float): Configuration for devices that support configurable readout times. + + Returns: + ScanReport + """ + super().__init__(**kwargs) + self.components = LamNIComponents(self) + self._baseline_readout_status = None + self.fovx = fovx + self.fovy = fovy + self.step = step + self.shift_x = shift_x + self.shift_y = shift_y + self.center_x = center_x + self.center_y = center_y + self.angle = angle + self.stitch_x = stitch_x + self.stitch_y = stitch_y + self.fov_circular = fov_circular + self.stitch_overlap = stitch_overlap + self.exp_time = exp_time + self.frames_per_trigger = frames_per_trigger + self.readout_time = readout_time + self.corridor_size = corridor_size + + self.update_scan_info( + exp_time=exp_time, frames_per_trigger=frames_per_trigger, readout_time=readout_time + ) + + @scan_hook + def prepare_scan(self): + """ + Prepare the scan. This can include any steps that need to be executed + before the scan is opened, such as preparing the positions (if not done already) + or setting up the devices. + """ + + positions = self.get_lamni_fermat_spiral_pos( + -np.abs(self.fovx / 2), + np.abs(self.fovx / 2), + -np.abs(self.fovy / 2), + np.abs(self.fovy / 2), + step=self.step, + spiral_type=0, + center=False, + center_x=self.center_x, + center_y=self.center_y, + angle=self.angle, + stitch_x=self.stitch_x, + stitch_y=self.stitch_y, + stitch_overlap=self.stitch_overlap, + shift_x=self.shift_x, + shift_y=self.shift_y, + fovx=self.fovx, + fovy=self.fovy, + fov_circular=self.fov_circular, + ) + + if len(positions) < self.MIN_POSITIONS: + raise ScanAbortion( + f"The number positions must exceed {self.MIN_POSITIONS}. Currently: {len(positions)}." + ) + + self.positions = self.components.optimize_trajectory( + positions, optimization_type="corridor", corridor_size=self.corridor_size + ) + + self.update_scan_info(num_points=len(self.positions), positions=self.positions) + + self.prepare_setup() + + self.actions.add_scan_report_instruction_device_progress(device="rt_positions") + + self._baseline_readout_status = self.actions.read_baseline_devices(wait=False) + + @scan_hook + def open_scan(self): + """ + Open the scan. + This step must call self.actions.open_scan() to ensure that a new scan is + opened. Make sure to prepare the scan metadata before, either in + prepare_scan() or in open_scan() itself and call self.update_scan_info(...) + to update the scan metadata if needed. + """ + self.actions.open_scan() + + @scan_hook + def stage(self): + """ + Stage the devices for the upcoming scan. The stage logic is typically + implemented on the device itself (i.e. by the device's stage method). + However, if there are any additional steps that need to be executed before + staging the devices, they can be implemented here. + """ + self.actions.stage_all_devices() + + @scan_hook + def pre_scan(self): + """ + Pre-scan steps to be executed before the main scan logic. + This is typically the last chance to prepare the devices before the core scan + logic is executed. For example, this is a good place to initialize time-criticial + devices, e.g. devices that have a short timeout. + The pre-scan logic is typically implemented on the device itself. + """ + self.actions.pre_scan_all_devices() + + @scan_hook + def scan_core(self): + """ + Core scan logic to be executed during the scan. + This is where the main scan logic should be implemented. + """ + + self.actions.kickoff(device="rt_positions") + + status = self.actions.complete(device="rt_positions", wait=False) + + while not status.done: + self.at_each_point() + time.sleep(1) + + @scan_hook + def at_each_point(self): + """ + Logic to be executed at each acquisition point during the scan. + """ + self.actions.read_monitored_devices() + + @scan_hook + def post_scan(self): + """ + Post-scan steps to be executed after the main scan logic. + """ + self.actions.complete_all_devices() + + @scan_hook + def unstage(self): + """Unstage the scan by executing post-scan steps.""" + self.actions.unstage_all_devices() + + @scan_hook + def close_scan(self): + """Close the scan.""" + if self._baseline_readout_status is not None: + self._baseline_readout_status.wait() + self.actions.close_scan() + self.actions.check_for_unchecked_statuses() + + @scan_hook + def on_exception(self, exception: Exception): + """ + Handle exceptions that occur during the scan. + This is a good place to implement any cleanup logic that needs to be executed in case of an exception, + such as returning the devices to a safe state or moving the motors back to their starting position. + """ + + ####################################################### + ######### Helper methods for the scan logic ########### + ####################################################### + + @staticmethod + def get_lamni_fermat_spiral_pos( + m1_start: float, + m1_stop: float, + m2_start: float, + m2_stop: float, + step: float = 1, + spiral_type: float = 0, + center: bool = False, + center_x: float = 0.0, + center_y: float = 0.0, + angle: float = 0.0, + stitch_x: float = 0.0, + stitch_y: float = 0.0, + stitch_overlap: float = 1.0, + shift_x: float = 0.0, + shift_y: float = 0.0, + fovx: float | None = None, + fovy: float | None = None, + fov_circular: float = 0.0, + ) -> np.ndarray: + """Generate positions for a LamNI Fermat spiral scan. + + Args: + m1_start (float): start position motor 1 + m1_stop (float): end position motor 1 + m2_start (float): start position motor 2 + m2_stop (float): end position motor 2 + step (float, optional): Step size. Defaults to 1. + spiral_type (float, optional): Angular offset in radians that determines the shape of the spiral. + A spiral with spiral_type=2 is the same as spiral_type=0. Defaults to 0. + center (bool, optional): Add a center point. Defaults to False. + + Returns: + np.ndarray: Array of positions for the Fermat spiral scan. + """ + positions = [] + phi = 2 * np.pi * ((1 + np.sqrt(5)) / 2.0) + spiral_type * np.pi + + start = int(not center) + + length_axis1 = np.abs(m1_stop - m1_start) + length_axis2 = np.abs(m2_stop - m2_start) + n_max = int(length_axis1 * length_axis2 * 3.2 / step / step) + + total_shift_x, total_shift_y = LamniFermatScan._compute_total_shift( + center_x=center_x, + center_y=center_y, + angle=angle, + stitch_x=stitch_x, + stitch_y=stitch_y, + stitch_overlap=stitch_overlap, + shift_x=shift_x, + shift_y=shift_y, + fovx=fovx, + fovy=fovy, + ) + + for ii in range(start, n_max): + radius = step * 0.57 * np.sqrt(ii) + x = radius * np.sin(ii * phi) + y = radius * np.cos(ii * phi) + if LamniFermatScan._lamni_check_pos_in_fov_range_and_circ_fov( + x=x, + y=y, + stitch_x=stitch_x, + stitch_y=stitch_y, + angle=angle, + fovx=fovx, + fovy=fovy, + stitch_overlap=stitch_overlap, + fov_circular=fov_circular, + ): + positions.extend([(x + total_shift_x * 1000, y + total_shift_y * 1000)]) + # for testing we just shift by center_i and prepare also the setup to center_i + return np.array(positions) + + @staticmethod + def _compute_total_shift( + center_x: float, + center_y: float, + angle: float, + stitch_x: float, + stitch_y: float, + stitch_overlap: float, + shift_x: float, + shift_y: float, + fovx: float | None, + fovy: float | None, + ) -> tuple[float, float]: + _shfitx, _shfity = LamNIComponents.lamni_compute_scan_center(center_x, center_y, angle) + x_stitch_shift, y_stitch_shift = LamniFermatScan._lamni_compute_stitch_center( + xcount=stitch_x, + ycount=stitch_y, + angle_deg=angle, + fovx=fovx, + fovy=fovy, + stitch_overlap=stitch_overlap, + ) + logger.info( + f"Total shift [mm] {_shfitx+x_stitch_shift/1000+shift_x}," + f" {_shfity+y_stitch_shift/1000+shift_y}" + ) + return ( + _shfitx + x_stitch_shift / 1000 + shift_x, + _shfity + y_stitch_shift / 1000 + shift_y, + ) + + @staticmethod + def _lamni_compute_stitch_center( + xcount: float, + ycount: float, + angle_deg: float, + fovx: float | None, + fovy: float | None, + stitch_overlap: float, + ) -> tuple[float, float]: + """ + Compute the stitch center in the stage coordinates based on the provided stitch counts and the rotation angle. + + Args: + xcount (float): Stitch count in x direction + ycount (float): Stitch count in y direction + angle_deg (float): Rotation angle in degrees + + Returns: + tuple: (shift_x, shift_y) in mm to be applied to the scan center in the stage coordinates + """ + if fovx is None or fovy is None: + raise ValueError("fovx and fovy must be provided to compute the stitch center.") + + alpha = angle_deg / 180 * np.pi + stage_x = xcount * (fovx - stitch_overlap) + stage_y = ycount * (fovy - stitch_overlap) + x_rot = np.cos(alpha) * stage_x - np.sin(alpha) * stage_y + y_rot = np.sin(alpha) * stage_x + np.cos(alpha) * stage_y + + return LamNIComponents.lamni_from_stage_coordinates(x_rot, y_rot) + + @staticmethod + def _lamni_check_pos_in_fov_range_and_circ_fov( + x: float, + y: float, + stitch_x: float, + stitch_y: float, + angle: float, + fovx: float | None, + fovy: float | None, + stitch_overlap: float, + fov_circular: float, + ) -> bool: + """ + Check if the given position is within the FOV range and circular FOV (if specified). + + Args: + x (float): x position in lamni coordinates (mm) + y (float): y position in lamni coordinates (mm) + + Returns: + bool: True if the position is within the FOV range and circular FOV, False otherwise. + """ + # this function checks if positions are reachable in a scan + # these x y intererometer positions are not shifted to the scan center + # so its purpose is to see if the position is reachable by the + # rotated piezo stage. For a scan these positions have to be shifted to + # the current scan center before starting the scan + if fovx is None or fovy is None: + raise ValueError("fovx and fovy must be provided to check LamNI scan positions.") + + stage_x, stage_y = LamNIComponents.lamni_to_stage_coordinates(x, y) + stage_x_with_stitch, stage_y_with_stitch = LamniFermatScan._lamni_compute_stitch_center( + xcount=stitch_x, + ycount=stitch_y, + angle_deg=angle, + fovx=fovx, + fovy=fovy, + stitch_overlap=stitch_overlap, + ) + stage_x_with_stitch, stage_y_with_stitch = LamNIComponents.lamni_to_stage_coordinates( + stage_x_with_stitch, stage_y_with_stitch + ) + + # piezo stage is currently rotated to stage_angle_deg in degrees + # rotate positions to the piezo stage system + alpha = (angle - 300 + 30.5) / 180 * np.pi + stage_x_rot = np.cos(alpha) * stage_x + np.sin(alpha) * stage_y + stage_y_rot = -np.sin(alpha) * stage_x + np.cos(alpha) * stage_y + + stage_x_rot_with_stitch = ( + np.cos(alpha) * stage_x_with_stitch + np.sin(alpha) * stage_y_with_stitch + ) + stage_y_rot_with_stitch = ( + -np.sin(alpha) * stage_x_with_stitch + np.cos(alpha) * stage_y_with_stitch + ) + + # FIXME: We are checking stage_x_rot vs fovy. This needs to be clarified if this is correct + # once LamNI is back in operation. We keep it like this for now. + return ( + np.abs(stage_x_rot) <= (fovy / 2) + and np.abs(stage_y_rot) <= (fovx / 2) + and ( + fov_circular == 0 + or ( + np.power((stage_x_rot_with_stitch + stage_x_rot), 2) + + np.power((stage_y_rot_with_stitch + stage_y_rot), 2) + ) + <= pow((fov_circular / 2), 2) + ) + ) + + def lamni_rotation(self, angle: float): + """ + Rotate LamNI to the specified angle. The rotation is only performed + if the current setpoint of the rotation stage is different from the requested angle. + + Args: + angle (float): Rotation angle in degrees + """ + # get last setpoint (cannot be based on pos get because they will deviate slightly) + lsamrot_current_setpoint = self.dev.lsamrot.user_setpoint.get() + if angle == lsamrot_current_setpoint: + logger.info("No rotation required") + return + + logger.info("Rotating to requested angle") + self.actions.add_scan_report_instruction_readback( + devices=["lsamrot"], start=[lsamrot_current_setpoint], stop=[angle] + ) + self.dev.lsamrot.set(angle).wait() + + def prepare_setup(self): + """ + Prepare the setup for the scan: + - Clear the trajectory generator of the RT controller to remove any previous positions. + - Rotate LamNI to the requested angle. + - Compute the total shift based on the center, stitch, and user-defined shifts, and + move to the new scan center using the interferometer feedback. + - Transfer the positions to the RT controller by adding them to the trajectory generator. + """ + self.dev.rtx.controller.clear_trajectory_generator() + self.lamni_rotation(self.angle) + total_shift_x, total_shift_y = LamniFermatScan._compute_total_shift( + center_x=self.center_x, + center_y=self.center_y, + angle=self.angle, + stitch_x=self.stitch_x, + stitch_y=self.stitch_y, + stitch_overlap=self.stitch_overlap, + shift_x=self.shift_x, + shift_y=self.shift_y, + fovx=self.fovx, + fovy=self.fovy, + ) + self.components.lamni_new_scan_center_interferometer(total_shift_x, total_shift_y) + + # Transfer the positions to the RT controller + self.dev.rtx.controller.add_pos_to_scan(self.positions.tolist()) diff --git a/csaxs_bec/scans/lamni_move_to_scan_center.py b/csaxs_bec/scans/lamni_move_to_scan_center.py new file mode 100644 index 0000000..bb9d7c6 --- /dev/null +++ b/csaxs_bec/scans/lamni_move_to_scan_center.py @@ -0,0 +1,115 @@ +""" +LamNI scan to move the interferometer to the computed scan center based on the provided shift and angle. + +Scan procedure: + - prepare_scan + - open_scan + - stage + - pre_scan + - scan_core + - at_each_point (optionally called by scan_core) + - post_scan + - unstage + - close_scan + - on_exception (called if any exception is raised during the scan) +""" + +from __future__ import annotations + +from typing import Annotated + +from bec_lib.scan_args import ScanArgument, Units +from bec_server.scan_server.scans.scan_base import ScanBase +from bec_server.scan_server.scans.scan_modifier import scan_hook + +from csaxs_bec.scans.lamni_components import LamNIComponents + + +class LamniMoveToScanCenter(ScanBase): + # Scan Type: Hardware triggered or software triggered? + # If the main trigger and readout logic is done within the at_each_point method in scan_core, choose SOFTWARE_TRIGGERED. + # If the main trigger and readout logic is implemented on a device that is simply kicked off in this scan, choose HARDWARE_TRIGGERED. + # This primarily serves as information for devices: The device may need to react differently if a software trigger is expected + # for every point. + scan_type = None + is_scan = False + + # Scan name: This is the name of the scan, e.g. "line_scan". This is used for display purposes and to identify the scan type in user interfaces. + # Choose a descriptive name that does not conflict with existing scan names. + # It must be a valid Python identifier, that is, it can only contain letters, numbers, and underscores, and must not start with a number. + scan_name = "lamni_move_to_scan_center" + + gui_config = {"Scan Parameters": ["shift_x", "shift_y", "angle"]} + + def __init__( + # fmt: off + self, + shift_x: Annotated[float, ScanArgument(display_name="Shift X", description="Shift x.", units=Units.mm)], + shift_y: Annotated[float, ScanArgument(display_name="Shift Y", description="Shift y.", units=Units.mm)], + angle: Annotated[float, ScanArgument(display_name="Angle", description="Angle.", units=Units.deg)], + **kwargs, + # fmt: on + ): + """ + LamNI scan to move the interferometer to the computed scan center based on the provided shift and angle. + + Args: + shift_x (float): Shift x. + shift_y (float): Shift y. + angle (float): Angle. + + Returns: + ScanReport + """ + super().__init__(**kwargs) + self.components = LamNIComponents(self) + self._baseline_readout_status = None + self.shift_x = shift_x + self.shift_y = shift_y + self.angle = angle + + self.update_scan_info() + + @scan_hook + def prepare_scan(self): ... + + @scan_hook + def open_scan(self): ... + + @scan_hook + def stage(self): ... + + @scan_hook + def pre_scan(self): ... + + @scan_hook + def scan_core(self): + """ + Core scan logic to be executed during the scan. + This is where the main scan logic should be implemented. + """ + center_x, center_y = self.components.lamni_compute_scan_center( + self.shift_x, self.shift_y, self.angle + ) + self.components.lamni_new_scan_center_interferometer(center_x, center_y) + + @scan_hook + def at_each_point(self): ... + + @scan_hook + def post_scan(self): ... + + @scan_hook + def unstage(self): ... + + @scan_hook + def close_scan(self): ... + + @scan_hook + def on_exception(self, exception: Exception): ... + + ####################################################### + ######### Helper methods for the scan logic ########### + ####################################################### + + # Implement scan-specific helper methods below. diff --git a/csaxs_bec/scans/omny_fermat_scan.py b/csaxs_bec/scans/omny_fermat_scan.py index b2f5bcb..5236f1f 100644 --- a/csaxs_bec/scans/omny_fermat_scan.py +++ b/csaxs_bec/scans/omny_fermat_scan.py @@ -1,205 +1,135 @@ """ -SCAN PLUGINS +OMNY Fermat's spiral scan -All new scans should be derived from ScanBase. ScanBase provides various methods that can be customized and overriden -but they are executed in a specific order: - -- self.initialize # initialize the class if needed -- self.read_scan_motors # used to retrieve the start position (and the relative position shift if needed) -- self.prepare_positions # prepare the positions for the scan. The preparation is split into multiple sub fuctions: - - self._calculate_positions # calculate the positions - - self._set_positions_offset # apply the previously retrieved scan position shift (if needed) - - self._check_limits # tests to ensure the limits won't be reached -- self.open_scan # send an open_scan message including the scan name, the number of points and the scan motor names -- self.stage # stage all devices for the upcoming acquisiton -- self.run_baseline_readings # read all devices to get a baseline for the upcoming scan -- self.scan_core # run a loop over all position - - self._at_each_point(ind, pos) # called at each position with the current index and the target positions as arguments -- self.finalize # clean up the scan, e.g. move back to the start position; wait everything to finish -- self.unstage # unstage all devices that have been staged before -- self.cleanup # send a close scan message and perform additional cleanups if needed +Scan procedure: + - prepare_scan + - open_scan + - stage + - pre_scan + - scan_core + - at_each_point (optionally called by scan_core) + - post_scan + - unstage + - close_scan + - on_exception (called if any exception is raised during the scan) """ +from __future__ import annotations + import time +from typing import Annotated import numpy as np -from bec_lib import bec_logger, messages -from bec_lib.endpoints import MessageEndpoints +from bec_lib import messages +from bec_lib.logger import bec_logger +from bec_lib.scan_args import DefaultArgType, ScanArgument, Units from bec_server.scan_server.errors import ScanAbortion -from bec_server.scan_server.scans import SyncFlyScanBase +from bec_server.scan_server.scans import MessageEndpoints +from bec_server.scan_server.scans.scan_base import ScanBase, ScanType +from bec_server.scan_server.scans.scan_modifier import scan_hook logger = bec_logger.logger -class OMNYFermatScan(SyncFlyScanBase): - scan_name = "omny_fermat_scan" - scan_report_hint = "table" - scan_type = "fly" - required_kwargs = ["fovx", "fovy", "exp_time", "step", "angle"] - arg_input = {} - arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None} +class OmnyFermatScan(ScanBase): + # Scan Type: Hardware triggered or software triggered? + # If the main trigger and readout logic is done within the at_each_point method in scan_core, choose SOFTWARE_TRIGGERED. + # If the main trigger and readout logic is implemented on a device that is simply kicked off in this scan, choose HARDWARE_TRIGGERED. + # This primarily serves as information for devices: The device may need to react differently if a software trigger is expected + # for every point. + scan_type = ScanType.HARDWARE_TRIGGERED + + # Scan name: This is the name of the scan, e.g. "line_scan". This is used for display purposes and to identify the scan type in user interfaces. + # Choose a descriptive name that does not conflict with existing scan names. + # It must be a valid Python identifier, that is, it can only contain letters, numbers, and underscores, and must not start with a number. + scan_name = "omny_fermat_scan_v4" + + gui_config = { + "Scan Parameters": [ + "fovx", + "fovy", + "cenx", + "ceny", + "step", + "zshift", + "angle", + "corridor_size", + ], + "Acquisition Parameters": ["exp_time", "frames_per_trigger", "readout_time"], + } def __init__( + # fmt: off self, - fovx: float, - fovy: float, - cenx: float, - ceny: float, - exp_time: float, - step: float, - zshift: float, - angle: float = None, - corridor_size: float = 3, - parameter: dict = None, - frames_per_trigger: int = 1, + fovx: Annotated[float, ScanArgument(display_name="Fovx", description="FOV in the piezo plane (i.e. piezo range).", units=Units.µm, gt=0, lt=200)], + fovy: Annotated[float, ScanArgument(display_name="Fovy", description="FOV in the piezo plane (i.e. piezo range).", units=Units.µm, gt=0, lt=100)], + cenx: Annotated[float, ScanArgument(display_name="Cenx", description="Center position in x.", units=Units.µm)], + ceny: Annotated[float, ScanArgument(display_name="Ceny", description="Center position in y.", units=Units.µm)], + step: Annotated[float, ScanArgument(display_name="Step", description="Step size.", units=Units.µm)], + zshift: Annotated[float, ScanArgument(display_name="Zshift", description="Shift in z.", units=Units.µm)], + angle: Annotated[float | None, ScanArgument(display_name="Angle", description="Rotation angle (will rotate first)", units=Units.deg)] = None, + corridor_size: Annotated[float, ScanArgument(display_name="Corridor Size", description="Corridor size for the corridor optimization. ", units=Units.µm)] = 3, + exp_time: DefaultArgType.ExposureTime = 0, + frames_per_trigger: DefaultArgType.FramesPerTrigger = 1, + readout_time: DefaultArgType.ReadoutTime = 0, **kwargs, + # fmt: on ): """ - An OMNY scan following Fermat's spiral. + OMNY Fermat's spiral scan Args: - fovx(float) [um]: Fov in the piezo plane (i.e. piezo range). Max 200 um - fovy(float) [um]: Fov in the piezo plane (i.e. piezo range). Max 100 um - cenx(float) [um]: center position in x. - ceny(float) [um]: center position in y. - exp_time(float) [s]: exposure time - frames_per_trigger:int: Number of burst frames per trigger, defaults to 1. - step(float) [um]: stepsize - zshift(float) [um]: shift in z - angle(float) [deg]: rotation angle (will rotate first) - corridor_size(float) [um]: corridor size for the corridor optimization. Default 3 um + fovx (float): FOV in the piezo plane (i.e. piezo range). + fovy (float): FOV in the piezo plane (i.e. piezo range). + cenx (float): Center position in x. + ceny (float): Center position in y. + step (float): Step size. + zshift (float): Shift in z. + angle (float | None): Rotation angle (will rotate first) + corridor_size (float): Corridor size for the corridor optimization. + exp_time (float): Exposure time in seconds + frames_per_trigger (int): Number of frames per trigger for devices that support configurable frame counts per trigger. + readout_time (float): Configuration for devices that support configurable readout times. Returns: - - Examples: - >>> scans.omny_fermat_scan(fovx=20, fovy=25, cenx=10, ceny=0, zshift=0, angle=0, step=2, exp_time=0.01) + ScanReport """ - - super().__init__( - parameter=parameter, exp_time=exp_time, frames_per_trigger=frames_per_trigger, **kwargs - ) - self.axis = [] + super().__init__(**kwargs) + self._baseline_readout_status = None self.fovx = fovx self.fovy = fovy self.cenx = cenx self.ceny = ceny - self.exp_time = exp_time self.step = step self.zshift = zshift self.angle = angle - self.optim_trajectory = "corridor" - self.optim_trajectory_corridor = corridor_size - if self.fovy > 100: - raise ScanAbortion("The FOV in y must be smaller than 100 um.") - if self.fovx > 200: - raise ScanAbortion("The FOV in x must be smaller than 200 um.") + self.corridor_size = corridor_size + self.exp_time = exp_time + self.frames_per_trigger = frames_per_trigger + self.readout_time = readout_time + if self.zshift > 100: logger.warning("The zshift is larger than 100 um. It will be limited to 100 um.") self.zshift = 100 - if self.zshift < -100: + elif self.zshift < -100: logger.warning("The zshift is smaller than -100 um. It will be limited to -100 um.") self.zshift = -100 + + self.update_scan_info( + exp_time=exp_time, frames_per_trigger=frames_per_trigger, readout_time=readout_time + ) + self.omny_rotation_status = None - def initialize(self): - self.scan_motors = [] - self.update_readout_priority() - - def _optimize_trajectory(self): - self.positions = self.optimize_corridor( - self.positions, corridor_size=self.optim_trajectory_corridor - ) - - @property - def monitor_sync(self): - return "rt_omny" - - def reverse_trajectory(self): + @scan_hook + def prepare_scan(self): """ - Reverse the trajectory. Every other scan should be reversed to - shorten the movement time. In order to keep the last state, even if the - server is restarted, the state is stored in a global variable in redis. + Prepare the scan. This can include any steps that need to be executed + before the scan is opened, such as preparing the positions (if not done already) + or setting up the devices. """ - msg = self.connector.get(MessageEndpoints.global_vars("reverse_omny_trajectory")) - if msg: - val = msg.content.get("value", False) - else: - val = False - self.connector.set( - MessageEndpoints.global_vars("reverse_omny_trajectory"), - messages.VariableMessage(value=(not val)), - ) - return val - def prepare_positions(self): - self._calculate_positions() - self._optimize_trajectory() - flip_axes = self.reverse_trajectory() - if flip_axes: - self.positions = np.flipud(self.positions) - - self.num_pos = len(self.positions) - self._check_min_positions() - - def _check_min_positions(self): - if self.num_pos < 20: - raise ScanAbortion( - f"The number of positions must exceed 20. Currently: {self.num_pos}." - ) - - def _prepare_setup(self): - yield from self.stubs.send_rpc_and_wait("rtx", "controller.clear_trajectory_generator") - yield from self.omny_rotation(self.angle) - - yield from self.stubs.send_rpc_and_wait("rty", "set", self.positions[0][1]) - - def _prepare_setup_part2(self): - if self.omny_rotation_status: - self.omny_rotation_status.wait() - - rtx_status = yield from self.stubs.set(device="rtx", value=self.positions[0][0], wait=False) - rtz_status = yield from self.stubs.set(device="rtz", value=self.positions[0][2], wait=False) - - yield from self.stubs.send_rpc_and_wait( - "rtx", "controller.laser_tracker_check_and_wait_for_signalstrength" - ) - - rtx_status.wait() - rtz_status.wait() - - yield from self._transfer_positions_to_omny() - yield from self.stubs.send_rpc_and_wait("osamx", "omny_osamx_to_scan_center", self.cenx) - - def omny_rotation(self, angle): - # get last setpoint (cannot be based on pos get because they will deviate slightly) - osamroy_current_setpoint = yield from self.stubs.send_rpc_and_wait( - "osamroy", "user_setpoint.get" - ) - if angle == osamroy_current_setpoint: - logger.info("No rotation required") - else: - logger.info("Rotating to requested angle") - yield from self.stubs.scan_report_instruction( - { - "readback": { - "RID": self.metadata["RID"], - "devices": ["osamroy"], - "start": [osamroy_current_setpoint], - "end": [angle], - } - } - ) - self.omny_rotation_status = yield from self.stubs.set( - device="osamroy", value=angle, wait=False - ) - - def _transfer_positions_to_omny(self): - yield from self.stubs.send_rpc_and_wait( - "rtx", "controller.add_pos_to_scan", self.positions.tolist() - ) - - def _calculate_positions(self): - self.positions = self.get_omny_fermat_spiral_pos( + positions = self.get_omny_fermat_spiral_pos( -np.abs(self.fovx / 2), np.abs(self.fovx / 2), -np.abs(self.fovy / 2), @@ -209,8 +139,146 @@ class OMNYFermatScan(SyncFlyScanBase): center=False, ) + if len(positions) < 20: + raise ScanAbortion( + f"The number of positions must exceed 20. Currently: {len(positions)}." + ) + + self.positions = self.components.optimize_trajectory( + positions=positions, corridor_size=self.corridor_size, optimization_type="corridor" + ) + flip_axes = self.reverse_trajectory() + if flip_axes: + self.positions = np.flipud(self.positions) + + self.update_scan_info(positions=self.positions, num_points=len(self.positions)) + + self.prepare_setup() + + self.actions.add_scan_report_instruction_device_progress(device="rt_positions") + self._baseline_readout_status = self.actions.read_baseline_devices(wait=False) + + @scan_hook + def open_scan(self): + """ + Open the scan. + This step must call self.actions.open_scan() to ensure that a new scan is + opened. Make sure to prepare the scan metadata before, either in + prepare_scan() or in open_scan() itself and call self.update_scan_info(...) + to update the scan metadata if needed. + """ + self.actions.open_scan() + + @scan_hook + def stage(self): + """ + Stage the devices for the upcoming scan. The stage logic is typically + implemented on the device itself (i.e. by the device's stage method). + However, if there are any additional steps that need to be executed before + staging the devices, they can be implemented here. + """ + self.actions.stage_all_devices() + + @scan_hook + def pre_scan(self): + """ + Pre-scan steps to be executed before the main scan logic. + This is typically the last chance to prepare the devices before the core scan + logic is executed. For example, this is a good place to initialize time-criticial + devices, e.g. devices that have a short timeout. + The pre-scan logic is typically implemented on the device itself. + """ + self.prepare_setup_part2() + self.actions.pre_scan_all_devices() + + @scan_hook + def scan_core(self): + """ + Core scan logic to be executed during the scan. + This is where the main scan logic should be implemented. + """ + + self.actions.kickoff(device="rt_positions") + + status = self.actions.complete(device="rt_positions", wait=False) + while not status.done: + self.at_each_point() + time.sleep(1) + + @scan_hook + def at_each_point(self): + """ + Logic to be executed at each acquisition point during the scan. + """ + self.actions.read_monitored_devices() + + @scan_hook + def post_scan(self): + """ + Post-scan steps to be executed after the main scan logic. + """ + move_status = None + if isinstance(self.positions, np.ndarray) and len(self.positions[-1]) == 3: + move_status = self.actions.set( + device=["rtx", "rty", "rtz"], value=self.positions[-1], wait=False + ) + + self.actions.complete_all_devices() + + if move_status is not None: + move_status.wait() + + @scan_hook + def unstage(self): + """Unstage the scan by executing post-scan steps.""" + self.actions.unstage_all_devices() + + @scan_hook + def close_scan(self): + """Close the scan.""" + if self._baseline_readout_status is not None: + self._baseline_readout_status.wait() + self.actions.close_scan() + self.actions.check_for_unchecked_statuses() + + @scan_hook + def on_exception(self, exception: Exception): + """ + Handle exceptions that occur during the scan. + This is a good place to implement any cleanup logic that needs to be executed in case of an exception, + such as returning the devices to a safe state or moving the motors back to their starting position. + """ + + ####################################################### + ######### Helper methods for the scan logic ########### + ####################################################### + + def reverse_trajectory(self): + """ + Reverse the trajectory. Every other scan should be reversed to + shorten the movement time. In order to keep the last state, even if the + server is restarted, the state is stored in a global variable in redis. + """ + msg = self.redis_connector.get(MessageEndpoints.global_vars("reverse_omny_trajectory")) + if msg: + val = msg.content.get("value", False) + else: + val = False + self.redis_connector.set( + MessageEndpoints.global_vars("reverse_omny_trajectory"), + messages.VariableMessage(value=(not val)), + ) + return val + def get_omny_fermat_spiral_pos( - self, m1_start, m1_stop, m2_start, m2_stop, step=1, spiral_type=0, center=False + self, + m1_start: float, + m1_stop: float, + m2_start: float, + m2_stop: float, + step: float = 1, + spiral_type: int = 0, + center: bool = False, ): """ Calculate positions for a Fermat spiral scan. @@ -262,43 +330,46 @@ class OMNYFermatScan(SyncFlyScanBase): positions.append(right_upper_corner) return np.array(positions) - def scan_core(self): - # use a device message to receive the scan number and - # scan ID before sending the message to the device server - yield from self.stubs.kickoff(device="rtx") - while True: - yield from self.stubs.kickoff(device="rt_positions") + def prepare_setup(self): + self.dev.rtx.controller.clear_trajectory_generator() - # start the readout loop of the flyer - status = yield from self.stubs.complete(device="rt_positions", wait=False) + if self.angle is not None: + self.omny_rotation(self.angle) - while not status.done: - yield from self.stubs.read(group="monitored", point_id=self.point_id) - self.point_id += 1 - time.sleep(1) - logger.debug("reading monitors") + self.actions.set(device="rty", value=self.positions[0][1]) - def move_to_start(self): - """return to the start position""" - # in omny, we need to move to the start position of the next scan, which is the end position of the current scan - # this method is called in finalize and overwrites the default move_to_start() - if isinstance(self.positions, np.ndarray) and len(self.positions[-1]) == 3: - yield from self.stubs.set(device=["rtx", "rty", "rtz"], value=self.positions[-1]) + def omny_rotation(self, angle: float): + """ + Rotate to the requested angle. + If the angle is the same as the current angle, no rotation will be performed. + + Args: + angle (float): Rotation angle in degrees. + """ + osamroy_current_setpoint = self.dev.osamroy.user_setpoint.get() + if angle == osamroy_current_setpoint: + logger.info("No rotation required.") return + logger.info("Rotating to requested angle") + self.actions.add_scan_report_instruction_readback( + devices=["osamroy"], start=[osamroy_current_setpoint], stop=[angle] + ) + self.omny_rotation_status = self.actions.set( + self.dev.osamroy.user_setpoint, angle, wait=False + ) - logger.warning("No positions found to return to start") + def prepare_setup_part2(self): + if self.omny_rotation_status is not None: + self.omny_rotation_status.wait() - def run(self): - self.initialize() - yield from self.read_scan_motors() - self.prepare_positions() - yield from self._prepare_setup() - yield from self.open_scan() - yield from self.stage() - yield from self.run_baseline_reading() - yield from self._prepare_setup_part2() - yield from self.pre_scan() - yield from self.scan_core() - yield from self.finalize() - yield from self.unstage() - yield from self.cleanup() + rt_move_status = self.actions.set( + device=["rtx", "rtz"], value=[self.positions[0][0], self.positions[0][2]], wait=False + ) + + self.dev.rtx.controller.laser_tracker_check_and_wait_for_signalstrength() + + rt_move_status.wait() + + self.dev.rtx.controller.add_pos_to_scan(self.positions.tolist()) + + self.dev.osamx.omny_osamx_to_scan_center(self.cenx) diff --git a/tests/tests_bec_ipython_client/test_fermat_position_warning.py b/tests/tests_bec_ipython_client/test_fermat_position_warning.py index 73ea8f7..36ac41a 100644 --- a/tests/tests_bec_ipython_client/test_fermat_position_warning.py +++ b/tests/tests_bec_ipython_client/test_fermat_position_warning.py @@ -3,15 +3,15 @@ _expected_fermat_position_count()/_fermat_min_positions() (lamni.py/ flomni.py), which predict a scan's point count before it ever reaches the scan server, by calling the exact same algorithm the real scan classes use (FlomniFermatScan.get_flomni_fermat_spiral_pos()/ -LamNIFermatScan.get_lamni_fermat_spiral_pos(), both now pure @staticmethods --- see csaxs_bec/scans/flomni_fermat_scan.py and LamNIFermatScan.py). +LamniFermatScan.get_lamni_fermat_spiral_pos(), both now pure @staticmethods +-- see csaxs_bec/scans/flomni_fermat_scan.py and lamni_fermat_scan.py). """ import csaxs_bec.bec_ipython_client.plugins.LamNI.lamni as lamni_module from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan -from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan +from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan class FakeClient: @@ -59,11 +59,11 @@ def make_lamni(): def test_flomni_min_positions_threshold_is_20(): - assert FlomniFermatScan._MIN_POSITIONS == 20 + assert FlomniFermatScan.MIN_POSITIONS == 20 def test_lamni_min_positions_threshold_is_20(): - assert LamNIFermatScan._MIN_POSITIONS == 20 + assert LamniFermatScan.MIN_POSITIONS == 20 def test_flomni_expected_position_count_below_threshold(): @@ -105,7 +105,7 @@ def test_lamni_expected_position_count_below_threshold(): assert count < lamni._fermat_min_positions() # matches calling the real scan-class algorithm directly, center tile, angle 0 - expected = LamNIFermatScan.get_lamni_fermat_spiral_pos( + expected = LamniFermatScan.get_lamni_fermat_spiral_pos( -2.5, 2.5, -2.5, @@ -115,7 +115,8 @@ def test_lamni_expected_position_count_below_threshold(): stitch_x=0, stitch_y=0, stitch_overlap=0.2, - fov_size=[5.0, 5.0], + fovx=5.0, + fovy=5.0, fov_circular=0.0, ) assert count == len(expected) diff --git a/tests/tests_bec_ipython_client/test_lamni_tomo_params_widget_math.py b/tests/tests_bec_ipython_client/test_lamni_tomo_params_widget_math.py index bb5878f..c424f75 100644 --- a/tests/tests_bec_ipython_client/test_lamni_tomo_params_widget_math.py +++ b/tests/tests_bec_ipython_client/test_lamni_tomo_params_widget_math.py @@ -30,7 +30,7 @@ from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import ( _lamni_set_tomo_fov_offset, _requested_to_stepsize, ) -from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan +from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan STEPSIZES = [10.0, 7.0, 25.0, 12.5] @@ -180,9 +180,9 @@ def test_compute_fermat_positions_lamni_matches_scan_class_below_threshold(): "lamni_stitch_y": 0, } count, min_positions = _compute_fermat_positions_lamni(params) - assert min_positions == LamNIFermatScan._MIN_POSITIONS + assert min_positions == LamniFermatScan.MIN_POSITIONS assert count < min_positions - expected = LamNIFermatScan.get_lamni_fermat_spiral_pos( + expected = LamniFermatScan.get_lamni_fermat_spiral_pos( -2.5, 2.5, -2.5, @@ -192,7 +192,8 @@ def test_compute_fermat_positions_lamni_matches_scan_class_below_threshold(): stitch_x=0, stitch_y=0, stitch_overlap=0.2, - fov_size=[5.0, 5.0], + fovx=5.0, + fovy=5.0, fov_circular=0.0, ) assert count == len(expected) @@ -242,4 +243,4 @@ def test_compute_fermat_positions_lamni_zero_step_is_safe(): {"lamni_piezo_range_x": 5.0, "lamni_piezo_range_y": 5.0, "tomo_shellstep": 0.0} ) assert count == 0 - assert min_positions == LamNIFermatScan._MIN_POSITIONS + assert min_positions == LamniFermatScan.MIN_POSITIONS diff --git a/tests/tests_bec_ipython_client/test_tomo_params_widget_math.py b/tests/tests_bec_ipython_client/test_tomo_params_widget_math.py index 7e504a7..1f1b1f6 100644 --- a/tests/tests_bec_ipython_client/test_tomo_params_widget_math.py +++ b/tests/tests_bec_ipython_client/test_tomo_params_widget_math.py @@ -53,7 +53,7 @@ def _flomni_reference(stepsize): def test_compute_fermat_positions_flomni_matches_scan_class_below_threshold(): params = {"fovx": 5.0, "fovy": 5.0, "tomo_shellstep": 2.0} count, min_positions = _compute_fermat_positions_flomni(params) - assert min_positions == FlomniFermatScan._MIN_POSITIONS + assert min_positions == FlomniFermatScan.MIN_POSITIONS assert count < min_positions expected = FlomniFermatScan.get_flomni_fermat_spiral_pos(-2.5, 2.5, -2.5, 2.5, step=2.0) assert count == len(expected) @@ -72,4 +72,4 @@ def test_compute_fermat_positions_flomni_zero_step_is_safe(): {"fovx": 5.0, "fovy": 5.0, "tomo_shellstep": 0.0} ) assert count == 0 - assert min_positions == FlomniFermatScan._MIN_POSITIONS + assert min_positions == FlomniFermatScan.MIN_POSITIONS diff --git a/tests/tests_scans/test_flomni_fermat_scan.py b/tests/tests_scans/test_flomni_fermat_scan.py deleted file mode 100644 index 6f6f395..0000000 --- a/tests/tests_scans/test_flomni_fermat_scan.py +++ /dev/null @@ -1,57 +0,0 @@ -from unittest import mock - -import pytest -from bec_server.device_server.tests.utils import DMMock -from bec_server.scan_server.tests.fixtures import * - -from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan - - -@pytest.fixture -def scan_request(scan_assembler): - flomni_request = scan_assembler( - FlomniFermatScan, - fovx=5, - fovy=5, - cenx=0.0, - ceny=0.0, - exp_time=0.1, - step=1, - zshift=0.0, - angle=0.0, - metadata={"RID": "1234"}, - ) - yield flomni_request - - -def test_flomni_fermat_scan(scan_request): - assert scan_request.fovx == 5 - assert scan_request.fovy == 5 - - -def test_flomni_rotation_no_rotation_required(scan_request): - with mock.patch.object(scan_request.stubs, "_get_result_from_status") as get_from_rpc_mock: - get_from_rpc_mock.return_value = 90 - with mock.patch.object(scan_request.stubs, "scan_report_instruction") as scan_report_mock: - with mock.patch.object(scan_request.stubs, "set") as set_mock: - list(scan_request.flomni_rotation(90)) - scan_report_mock.assert_not_called() - assert not set_mock.called - - -def test_flomni_rotation_rotation_required(scan_request): - with mock.patch.object(scan_request.stubs, "_get_result_from_status", return_value=0): - with mock.patch.object(scan_request.stubs, "scan_report_instruction") as scan_report_mock: - with mock.patch.object(scan_request.stubs, "set") as set_mock: - list(scan_request.flomni_rotation(90)) - scan_report_mock.assert_called_once_with( - { - "readback": { - "RID": scan_request.metadata["RID"], - "devices": ["fsamroy"], - "start": [0], - "end": [90], - } - } - ) - set_mock.assert_called_once_with(device="fsamroy", value=90, wait=False) diff --git a/tests/tests_scans/test_lamni_fermat_scan.py b/tests/tests_scans/test_lamni_fermat_scan.py deleted file mode 100644 index 4985519..0000000 --- a/tests/tests_scans/test_lamni_fermat_scan.py +++ /dev/null @@ -1,422 +0,0 @@ -from unittest import mock - -import numpy as np -import pytest -from bec_lib import messages -from bec_server.device_server.tests.utils import DMMock -from bec_server.scan_server.errors import ScanAbortion -from bec_server.scan_server.tests.fixtures import ( - DeviceMockType, - DMMock, - ScanStubStatusMock, - connector_mock, - instruction_handler_mock, - scan_assembler, -) - -from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan - - -@pytest.fixture -def device_manager_mock(): - device_manager = DMMock() - device_manager.add_device("lsamx") - device_manager.devices["lsamx"]._config["userParameter"] = {"center": 8.1} - device_manager.add_device("lsamy") - device_manager.devices["lsamy"]._config["userParameter"] = {"center": 10} - device_manager.add_device("samx") - device_manager.devices["samx"].read_buffer = {"value": 0} - device_manager.add_device("samy") - device_manager.devices["samy"].read_buffer = {"value": 0} - device_manager.add_device("bpm4i", dev_type=DeviceMockType.SIGNAL, readout_priority="monitored") - yield device_manager - - -@pytest.mark.parametrize( - "scan_msg,reference_scan_list", - [ - ( - messages.ScanQueueMessage( - scan_type="lamni_fermat_scan", - parameter={ - "args": {}, - "kwargs": { - "fov_size": [5], - "exp_time": 0.1, - "step": 2, - "angle": 10, - "scan_type": "step", - }, - }, - queue="primary", - metadata={"RID": "1234"}, - ), - [ - None, - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rtx", - action="rpc", - parameter={ - "device": "rtx", - "func": "controller.clear_trajectory_generator", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="lsamrot", - action="rpc", - parameter={ - "device": "lsamrot", - "func": "user_setpoint.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={"readout_priority": "monitored", "RID": "1234"}, - device=None, - action="scan_report_instruction", - parameter={ - "readback": { - "RID": "1234", - "devices": ["lsamrot"], - "start": [0], - "end": [10], - } - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="lsamrot", - action="set", - parameter={"value": 10}, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rtx", - action="rpc", - parameter={ - "device": "rtx", - "func": "controller.feedback_disable", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rtx", - action="rpc", - parameter={ - "device": "rtx", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rty", - action="rpc", - parameter={ - "device": "rty", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="lsamx", - action="rpc", - parameter={ - "device": "lsamx", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="lsamy", - action="rpc", - parameter={ - "device": "lsamy", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rtx", - action="rpc", - parameter={ - "device": "rtx", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rty", - action="rpc", - parameter={ - "device": "rty", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="lsamrot", - action="rpc", - parameter={ - "device": "lsamrot", - "func": "readback.get", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rtx", - action="rpc", - parameter={ - "device": "rtx", - "func": "controller.feedback_enable_without_reset", - "rpc_id": "rpc_id", - "args": (), - "kwargs": {}, - }, - ), - messages.DeviceInstructionMessage( - metadata={"readout_priority": "monitored", "RID": "1234"}, - device=None, - action="scan_report_instruction", - parameter={"device_progress": ["rt_positions"]}, - ), - messages.DeviceInstructionMessage( - metadata={"readout_priority": "monitored", "RID": "1234"}, - device=None, - action="open_scan", - parameter={ - "readout_priority": { - "monitored": [], - "baseline": [], - "on_request": [], - "async": [], - }, - "num_points": 2, - "positions": [ - [-0.7700589354581364, -0.8406005210092851], - [1.3681828686580249, 2.1508313829565293], - ], - "scan_name": "lamni_fermat_scan", - "scan_type": "step", - }, - ), - messages.DeviceInstructionMessage( - metadata={"device_instr_id": "diid"}, - device=["bpm4i", "lsamx", "lsamy", "samx", "samy"], - action="stage", - parameter={}, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "baseline", - "RID": "1234", - "device_instr_id": "diid", - }, - device=["lsamx", "lsamy", "samx", "samy"], - action="read", - parameter={}, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device=["bpm4i", "lsamx", "lsamy", "samx", "samy"], - action="pre_scan", - parameter={}, - ), - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "device_instr_id": "diid", - }, - device="rt_positions", - action="kickoff", - parameter={"configure": {}}, - ), - None, - messages.DeviceInstructionMessage( - metadata={ - "readout_priority": "monitored", - "RID": "1234", - "point_id": 0, - "device_instr_id": "diid", - }, - device=["bpm4i"], - action="read", - parameter={"group": "monitored"}, - ), - None, - messages.DeviceInstructionMessage( - metadata={"device_instr_id": "diid"}, - device=["bpm4i", "lsamx", "lsamy", "samx", "samy"], - action="unstage", - parameter={}, - ), - messages.DeviceInstructionMessage( - metadata={"readout_priority": "monitored", "RID": "1234"}, - device=None, - action="close_scan", - parameter={}, - ), - ], - ) - ], -) -def test_LamNIFermatScan(scan_msg, reference_scan_list, scan_assembler, ScanStubStatusMock): - scan = scan_assembler( - LamNIFermatScan, - parameter=scan_msg.content.get("parameter"), - metadata=scan_msg.metadata, - **scan_msg.content["parameter"]["kwargs"], - ) - - def fake_done(): - yield False - yield True - - def fake_complete(*args, **kwargs): - yield None - return ScanStubStatusMock(done_func=fake_done) - - with mock.patch.object(scan.stubs, "_get_result_from_status", return_value=0): - with mock.patch.object(scan, "_check_min_positions") as check_min_pos: - with mock.patch.object(scan.stubs, "complete", side_effect=fake_complete): - scan_instructions = list(scan.run()) - check_min_pos.assert_called_once() - - for ii, instr in enumerate(scan_instructions): - if instr is None: - continue - if instr.metadata.get("scan_id") is not None: - instr.metadata["scan_id"] = "scan_id" - if instr.metadata.get("RID") is not None: - instr.metadata["RID"] = scan.metadata.get("RID") - if instr.metadata.get("device_instr_id") is not None: - instr.metadata["device_instr_id"] = "diid" - if instr.content["action"] == "rpc": - instr.content["parameter"]["rpc_id"] = "rpc_id" - if instr.content["parameter"].get("value"): - assert np.isclose( - instr.content["parameter"].get("value"), - reference_scan_list[ii].content["parameter"].get("value"), - ) - instr.content["parameter"]["value"] = reference_scan_list[ii].content[ - "parameter" - ]["value"] - if instr.content["parameter"].get("positions"): - assert np.isclose( - instr.content["parameter"].get("positions"), - reference_scan_list[ii].content["parameter"].get("positions"), - ).all() - instr.content["parameter"]["positions"] = reference_scan_list[ii].content[ - "parameter" - ]["positions"] - assert scan_instructions == reference_scan_list - - -def test_LamNIFermatScan_min_positions(scan_assembler): - scan_msg = messages.ScanQueueMessage( - scan_type="lamni_fermat_scan", - parameter={ - "args": {}, - "kwargs": { - "fov_size": [5], - "exp_time": 0.1, - "step": 2, - "angle": 10, - "scan_type": "step", - }, - }, - queue="primary", - metadata={"RID": "1234"}, - ) - - scan = scan_assembler( - LamNIFermatScan, - parameter=scan_msg.content.get("parameter"), - metadata=scan_msg.metadata, - **scan_msg.content["parameter"]["kwargs"], - ) - with pytest.raises(ScanAbortion): - instructions = list(scan.run())