f

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"""

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)