diff --git a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py index 3e7fea9..033ecc7 100644 --- a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py +++ b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py @@ -734,9 +734,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/scans/omny_fermat_scan_v4.py b/csaxs_bec/scans/omny_fermat_scan_v4.py index 7b3678f..4ed21d7 100644 --- a/csaxs_bec/scans/omny_fermat_scan_v4.py +++ b/csaxs_bec/scans/omny_fermat_scan_v4.py @@ -16,13 +16,15 @@ Scan procedure: from __future__ import annotations +import time from typing import Annotated import numpy as np 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.scans import MessageEndpoints, position_generators +from bec_server.scan_server.errors import ScanAbortion +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 @@ -127,7 +129,7 @@ class OmnyFermatScanV4(ScanBase): or setting up the devices. """ - 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), @@ -137,35 +139,23 @@ class OmnyFermatScanV4(ScanBase): center=False, ) - # TODO: If necessary, prepare the positions for the scan here and update the scan - # info with the updated num_points and num_monitored_readouts using self.update_scan_info(). - # Take the following example of a hexagonal scan as a reference: + if len(positions) < 20: + raise ScanAbortion( + f"The number of positions must exceed 20. Currently: {len(positions)}." + ) - # self.positions = position_generators.hex_grid_2d( - # [ - # (self.start_motor1, self.stop_motor1, self.step_motor1), - # (self.start_motor2, self.stop_motor2, self.step_motor2), - # ], - # snaked=self.snaked, - # ) - - # if self.relative: - # self.start_positions = self.components.get_start_positions(self.motors) - # self.positions += self.start_positions - - # self.components.check_limits(self.motors, self.positions) - - # self.update_scan_info( - # positions=self.positions, - # num_points=len(self.positions), - # num_monitored_readouts=len(self.positions) * self.burst_at_each_point, - # ) - - # Delete the comment if not needed. - - self.actions.add_scan_report_instruction_scan_progress( - points=self.scan_info.num_monitored_readouts, show_table=False + 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 @@ -198,6 +188,7 @@ class OmnyFermatScanV4(ScanBase): 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 @@ -207,24 +198,36 @@ class OmnyFermatScanV4(ScanBase): This is where the main scan logic should be implemented. """ - # TODO: Implement the core scan logic here. If applicable, we recommend to call - # self.at_each_point() at each acquisition point in the scan to allow scan modifiers - # to extend the behavior of the scan at each point. - # Delete the comment if not needed. + 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.""" @@ -326,3 +329,47 @@ class OmnyFermatScanV4(ScanBase): positions.append(left_lower_corner) positions.append(right_upper_corner) return np.array(positions) + + def prepare_setup(self): + self.dev.rtx.controller.clear_trajectory_generator() + + if self.angle is not None: + self.omny_rotation(self.angle) + + self.actions.set(device="rty", value=self.positions[0][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 + ) + + def prepare_setup_part2(self): + if self.omny_rotation_status is not None: + self.omny_rotation_status.wait() + + 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)