"""
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, messages
from bec_lib.endpoints import MessageEndpoints
from bec_server.scan_server.errors import ScanAbortion
from bec_server.scan_server.scans import SyncFlyScanBase
logger = bec_logger.logger
[docs]
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}
def __init__(
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,
**kwargs,
):
"""
An OMNY scan following Fermat's spiral.
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
Returns:
Examples:
>>> scans.omny_fermat_scan(fovx=20, fovy=25, cenx=10, ceny=0, zshift=0, angle=0, step=2, exp_time=0.01)
"""
super().__init__(
parameter=parameter, exp_time=exp_time, frames_per_trigger=frames_per_trigger, **kwargs
)
self.axis = []
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.")
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.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"
[docs]
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_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
[docs]
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(
-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,
)
[docs]
def get_omny_fermat_spiral_pos(
self, m1_start, m1_stop, m2_start, m2_stop, step=1, spiral_type=0, center=False
):
"""
Calculate positions for a Fermat spiral scan.
Args:
m1_start(float): start position in m1
m1_stop(float): stop position in m1
m2_start(float): start position in m2
m2_stop(float): stop position in m2
step(float): stepsize
spiral_type(int): 0 for traditional Fermat spiral
center(bool): whether to include the center position
Returns:
positions(array): positions
"""
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)
z_pos = self.zshift
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)
positions.append([x + self.cenx, y + self.ceny, z_pos])
left_lower_corner = [
min(m1_start, m1_stop) + self.cenx,
min(m2_start, m2_stop) + self.ceny,
z_pos,
]
right_upper_corner = [
max(m1_start, m1_stop) + self.cenx,
max(m2_start, m2_stop) + self.ceny,
z_pos,
]
positions.append(left_lower_corner)
positions.append(right_upper_corner)
return np.array(positions)
[docs]
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")
# 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")
[docs]
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])
return
logger.warning("No positions found to return to start")
[docs]
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()