added new files for alignment, tools and laser

This commit is contained in:
Holler Mirko
2024-11-05 15:06:35 +01:00
committed by wakonig_k
parent 90449d22ee
commit f0df487131
6 changed files with 338 additions and 813 deletions
@@ -0,0 +1 @@
from .omny import OMNY
@@ -13,6 +13,8 @@ from typeguard import typechecked
from csaxs_bec.bec_ipython_client.plugins.cSAXS import cSAXSBeamlineChecks
from csaxs_bec.bec_ipython_client.plugins.omny.omny_optics_mixin import OMNYOpticsMixin
from csaxs_bec.bec_ipython_client.plugins.omny.omny_alignment import OMNYAlignmentMixin
from csaxs_bec.bec_ipython_client.plugins.omny.omny_sample_transfer_mixin import OMNYSampleTransferMixin
#from csaxs_bec.bec_ipython_client.plugins.flomni.x_ray_eye_align import XrayEyeAlign
logger = bec_logger.logger
@@ -428,785 +430,12 @@ class OMNYInitStagesMixin:
else:
return False
class OMNYSampleTransferMixin:
def _otransfer_gripper_safe_xray_in_operation(self):
#if(!_ogalil_all_axes_on_all_controllers_are_referenced())
#{
# printf("Not all axes are referenced. Aborting.\n")
# exit
#}
#_otransfer_gripper_to_park_z
#}'
class FlomniSampleTransferMixin:
def ensure_osa_back(self):
dev.fosaz.limits = [-12.6, -12.4]
umv(dev.fosaz, -12.5)
curtain_is_triggered = dev.fheater.controller.fosaz_light_curtain_is_triggered()
if not curtain_is_triggered:
raise FlomniError("Fosaz did not reach light curtain")
def move_fheater_up(self):
self.ensure_fheater_up()
def ensure_fheater_up(self):
axis_id = dev.fheater._config["deviceConfig"].get("axis_Id")
axis_id_numeric = self.axis_id_to_numeric(axis_id)
low, high = dev.fheater.controller.get_motor_limit_switch(axis_id)
if high:
raise FlomniError("fheater in high limit. How did we get here?? Aborting.")
if not low:
self.ensure_osa_back()
if dev.fheater.readback.get() < -0.2:
umv(dev.fheater, -0.2)
dev.fheater.controller.drive_axis_to_limit(axis_id_numeric, "reverse")
def move_fheater_down(self):
axis_id = dev.fheater._config["deviceConfig"].get("axis_Id")
axis_id_numeric = self.axis_id_to_numeric(axis_id)
self.ensure_osa_back()
fsamx_in = dev.fsamx.user_parameter.get("in")
if not np.isclose(dev.fsamx.readback.get(), fsamx_in, 0.2):
raise FlomniError("fsamx not in position. Aborting.")
fheater_in = dev.fheater.user_parameter.get("in")
umv(dev.fheater, fheater_in)
def ensure_gripper_up(self):
axis_id = dev.ftransy._config["deviceConfig"].get("axis_Id")
axis_id_numeric = self.axis_id_to_numeric(axis_id)
low, high = dev.ftransy.controller.get_motor_limit_switch(axis_id)
if low:
raise FlomniError("Ftransy in low limit. How did we get here?? Aborting.")
if high:
return
if dev.ftransy.readback.get() < -0.5:
umv(dev.ftransy, -0.5)
dev.ftransy.controller.drive_axis_to_limit(axis_id_numeric, "forward")
def check_tray_in(self):
axis_id = dev.ftray._config["deviceConfig"].get("axis_Id")
low, high = dev.ftray.controller.get_motor_limit_switch(axis_id)
if high:
raise FlomniError("Ftray is in the 'OUT' position. Aborting.")
if not low:
raise FlomniError("Ftray is not at the 'IN' position. Aborting.")
def ftransfer_flomni_stage_in(self):
sample_in_position = bool(float(dev.flomni_samples.sample_placed.sample0.get()))
if not sample_in_position:
raise FlomniError("There is no sample in the sample stage. Aborting.")
self.reset_correction()
dev.rtx.controller.feedback_disable()
self.ensure_fheater_up()
self.ensure_gripper_up()
self.check_tray_in()
fsamx_in = dev.fsamx.user_parameter.get("in")
umv(dev.fsamx, fsamx_in)
dev.fsamx.limits = [fsamx_in - 0.4, fsamx_in + 0.4]
def laser_tracker_show_all(self):
dev.rtx.controller.laser_tracker_show_all()
def laser_tracker_on(self):
dev.rtx.controller.laser_tracker_on()
time.sleep(0.2)
self._laser_tracker_check_signalstrength()
def laser_tracker_off(self):
dev.rtx.controller.laser_tracker_off()
def show_signal_strength_interferometer(self):
dev.rtx.controller.show_signal_strength_interferometer()
def rt_feedback_disable(self):
self.device_manager.devices.rtx.controller.feedback_disable()
def rt_feedback_enable_with_reset(self):
self.device_manager.devices.rtx.controller.feedback_enable_with_reset()
self.rt_feedback_status()
def rt_feedback_enable_without_reset(self):
self.device_manager.devices.rtx.controller.feedback_enable_without_reset()
self.rt_feedback_status()
def rt_feedback_status(self):
feedback_status = self.device_manager.devices.rtx.controller.feedback_is_running()
if feedback_status == True:
print("The rt feedback is \x1b[92mrunning\x1b[0m.")
else:
print("The rt feedback is \x1b[91mNOT\x1b[0m running.")
def lights_off(self):
self.device_manager.devices.fsamx.controller.lights_off()
def lights_on(self):
self.device_manager.devices.fsamx.controller.lights_on()
def ftransfer_flomni_stage_out(self):
target_pos = -162
if np.isclose(dev.fsamx.readback.get(), target_pos, 0.01):
return
umv(dev.fsamroy, 0)
self.rt_feedback_disable()
self.ensure_fheater_up()
self.ensure_gripper_up()
self.check_tray_in()
self.laser_tracker_off()
time.sleep(0.05)
fsamy_in = dev.fsamy.user_parameter.get("in")
if fsamy_in is None:
raise FlomniError(
"Could not find an 'IN' position for fsamy. Please check your config."
)
umv(dev.fsamy, fsamy_in)
time.sleep(0.05)
self.laser_tracker_on()
time.sleep(0.05)
self.laser_tracker_off()
time.sleep(0.05)
self.drive_axis_to_limit(dev.fsamx, "forward")
dev.fsamx.limits = [-162, 0]
dev.fsamx.controller.socket_put_confirmed("axspeed[4]=25*stppermm[4]")
umv(dev.fsamx, target_pos)
def check_sensor_connected(self):
sensor_voltage_target = dev.ftransy.user_parameter.get("sensor_voltage")
sensor_voltage = float(dev.ftransy.controller.socket_put_and_receive("MG@AN[1]").strip())
if not np.isclose(sensor_voltage, sensor_voltage_target, 0.5):
raise FlomniError(f"Sensor voltage is {sensor_voltage}, indicates an error. Aborting.")
def ftransfer_get_sample(self, position: int):
self.check_position_is_valid(position)
self.check_tray_in()
self.check_sensor_connected()
sample_in_gripper = bool(float(dev.flomni_samples.sample_in_gripper.get()))
if sample_in_gripper:
raise FlomniError(
"The gripper does carry a sample. Cannot proceed getting another sample."
)
sample_signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
sample_in_position = bool(float(sample_signal.get()))
if not sample_in_position:
raise FlomniError(f"The planned pick position [{position}] does not have a sample.")
user_input = input(
"Please confirm that there is currently no sample in the gripper. It would be dropped!"
" [y/n]"
)
if user_input == "y":
print("good then")
else:
print("Stopping.")
raise FlomniError("The sample transfer was manually aborted.")
self.ftransfer_gripper_move(position)
self.ftransfer_controller_enable_mount_mode()
if position == 0:
sample_height = -45 + dev.fsamy.user_parameter.get("in")
else:
sample_height = -17.5
dev.ftransy.controller.socket_put_confirmed(f"getaprch={sample_height:.1f}")
dev.ftransy.controller.socket_put_confirmed("XQ#GRGET,3")
print("The unmount process started.")
time.sleep(1)
while True:
in_progress = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntprgs").strip())
)
if not in_progress:
break
self.ftransfer_confirm()
time.sleep(1)
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
self.flomni_modify_storage_non_interactive(100, 1, signal_name.get())
self.flomni_modify_storage_non_interactive(position, 0, "-")
def ftransfer_show_all(self):
dev.flomni_samples.show_all()
def ftransfer_put_sample(self, position: int):
self.check_position_is_valid(position)
self.check_tray_in()
self.check_sensor_connected()
sample_in_gripper = bool(float(dev.flomni_samples.sample_in_gripper.get()))
if not sample_in_gripper:
raise FlomniError("The gripper does not carry a sample.")
sample_signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
sample_in_position = bool(float(sample_signal.get()))
if sample_in_position:
raise FlomniError(f"The planned put position [{position}] already has a sample.")
self.ftransfer_gripper_move(position)
self.ftransfer_controller_enable_mount_mode()
if position == 0:
sample_height = -45 + dev.fsamy.user_parameter.get("in")
else:
sample_height = -17.5
dev.ftransy.controller.socket_put_confirmed(f"mntaprch={sample_height:.1f}")
dev.ftransy.controller.socket_put_confirmed("XQ#GRPUT,3")
print("The mount process started.")
time.sleep(1)
while True:
in_progress = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntprgs").strip())
)
if not in_progress:
break
self.ftransfer_confirm()
time.sleep(1)
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
sample_name = dev.flomni_samples.sample_in_gripper.get()
self.flomni_modify_storage_non_interactive(100, 0, "-")
self.flomni_modify_storage_non_interactive(position, 1, sample_name)
# TODO: flomni_stage_in if position == 0
# bec.queue.next_dataset_number += 1
def sample_get_name(self, position: int = 0) -> str:
"""
Get the name of the sample currently in the given position.
"""
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
return signal_name.get()
def ftransfer_sample_change(self, new_sample_position: int):
self.check_tray_in()
sample_in_gripper = dev.flomni_samples.sample_in_gripper.get()
if sample_in_gripper:
raise FlomniError("There is already a sample in the gripper. Aborting.")
self.check_position_is_valid(new_sample_position)
sample_placed = getattr(
dev.flomni_samples.sample_placed, f"sample{new_sample_position}"
).get()
if not sample_placed:
raise FlomniError(
f"There is currently no sample in position [{new_sample_position}]. Aborting."
)
sample_in_sample_stage = dev.flomni_samples.sample_placed.sample0.get()
if sample_in_sample_stage:
# find a new home for the sample...
empty_slots = []
for name, val in dev.flomni_samples.read().items():
if "flomni_samples_sample_placed_sample" not in name:
continue
if val.get("value") == 0:
empty_slots.append(int(name.split("flomni_samples_sample_placed_sample")[1]))
if not empty_slots:
raise FlomniError("There are no empty slots available. Aborting.")
print(f"The following slots are empty: {empty_slots}.")
while True:
user_input = input(f"Where shall I put the sample? Default: [{empty_slots[0]}]")
try:
user_input = int(user_input)
if user_input not in empty_slots:
raise ValueError
break
except ValueError:
print("Please specify a valid number.")
continue
self.check_position_is_valid(user_input)
self.ftransfer_get_sample(0)
self.ftransfer_put_sample(user_input)
self.ftransfer_get_sample(new_sample_position)
self.ftransfer_put_sample(0)
def ftransfer_modify_storage(self, position: int, used: int):
if used:
name = input("What's the name of this sample? ")
else:
name = "-"
self.flomni_modify_storage_non_interactive(position, used, name)
def flomni_modify_storage_non_interactive(self, position: int, used: int, name: str):
if position == 100:
dev.flomni_samples.sample_in_gripper.set(used)
dev.flomni_samples.sample_in_gripper_name.set(name)
else:
signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
signal.set(used)
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
signal_name.set(name)
def check_position_is_valid(self, position: int):
if 0 <= position < 21:
return
raise FlomniError(
f"The given position number [{position}] is not in the valid range of 0-21. "
)
def ftransfer_controller_enable_mount_mode(self):
dev.ftransy.controller.socket_put_confirmed("XQ#MNTMODE")
time.sleep(0.5)
if not self.ftransfer_controller_in_mount_mode():
raise FlomniError("System not switched to mount mode. Aborting.")
def ftransfer_controller_disable_mount_mode(self):
dev.ftransy.controller.socket_put_confirmed("XQ#POSMODE")
time.sleep(0.5)
if self.ftransfer_controller_in_mount_mode():
raise FlomniError("System is still in mount mode. Aborting.")
def ftransfer_controller_in_mount_mode(self) -> bool:
in_mount_mode = bool(
float(dev.ftransy.controller.socket_put_and_receive("MG mntmod").strip())
)
return in_mount_mode
def ftransfer_confirm(self):
confirm = int(float(dev.ftransy.controller.socket_put_and_receive("MG confirm").strip()))
if confirm != -1:
return
user_input = input("All OK? Continue? [y/n]")
if user_input == "y":
print("good then")
dev.ftransy.controller.socket_put_confirmed("confirm=1")
else:
print("Stopping.")
return
def ftransfer_gripper_is_open(self) -> bool:
status = bool(float(dev.ftransy.controller.socket_put_and_receive("MG @OUT[9]").strip()))
return status
def ftransfer_gripper_open(self):
sample_in_gripper = dev.flomni_samples.sample_in_gripper.get()
if sample_in_gripper:
raise FlomniError(
"Cannot open gripper. There is still a sample in the gripper! Aborting."
)
if not self.ftransfer_gripper_is_open():
dev.ftransy.controller.socket_put_confirmed("XQ#GROPEN,4")
def ftransfer_gripper_close(self):
if self.ftransfer_gripper_is_open():
dev.ftransy.controller.socket_put_confirmed("XQ#GRCLOS,4")
def ftransfer_gripper_move(self, position: int):
self.check_position_is_valid(position)
self._ftransfer_shiftx = -0.2
self._ftransfer_shiftz = -0.5
fsamx_pos = dev.fsamx.readback.get()
if position == 0 and fsamx_pos > -160:
user_input = input(
"May the flomni stage be moved out for the sample change? Feedback will be disabled"
" and alignment will be lost! [y/n]"
)
if user_input == "y":
print("good then")
self.ftransfer_flomni_stage_out()
else:
print("Stopping.")
return
self.ensure_gripper_up()
self.check_tray_in()
if position == 0:
umv(dev.ftransx, 10.715 + 0.2, dev.ftransz, 3.5950)
if position == 1:
umv(
dev.ftransx,
41.900 + self._ftransfer_shiftx,
dev.ftransz,
74.7500 + self._ftransfer_shiftz,
)
if position == 2:
umv(
dev.ftransx,
31.900 + self._ftransfer_shiftx,
dev.ftransz,
74.7625 + self._ftransfer_shiftz,
)
if position == 3:
umv(
dev.ftransx,
21.900 + self._ftransfer_shiftx,
dev.ftransz,
74.7750 + self._ftransfer_shiftz,
)
if position == 4:
umv(
dev.ftransx,
11.900 + self._ftransfer_shiftx,
dev.ftransz,
74.7875 + self._ftransfer_shiftz,
)
if position == 5:
umv(
dev.ftransx,
1.9000 + self._ftransfer_shiftx,
dev.ftransz,
74.8000 + self._ftransfer_shiftz,
)
if position == 6:
umv(
dev.ftransx,
41.900 + self._ftransfer_shiftx,
dev.ftransz,
89.7500 + self._ftransfer_shiftz,
)
if position == 7:
umv(
dev.ftransx,
31.900 + self._ftransfer_shiftx,
dev.ftransz,
89.7625 + self._ftransfer_shiftz,
)
if position == 8:
umv(
dev.ftransx,
21.900 + self._ftransfer_shiftx,
dev.ftransz,
89.7750 + self._ftransfer_shiftz,
)
if position == 9:
umv(
dev.ftransx,
11.900 + self._ftransfer_shiftx,
dev.ftransz,
89.7875 + self._ftransfer_shiftz,
)
if position == 10:
umv(
dev.ftransx,
1.900 + self._ftransfer_shiftx,
dev.ftransz,
89.8000 + self._ftransfer_shiftz,
)
if position == 11:
umv(
dev.ftransx,
41.95 + self._ftransfer_shiftx,
dev.ftransz,
124.75 + self._ftransfer_shiftz,
)
if position == 12:
umv(
dev.ftransx,
31.95 + self._ftransfer_shiftx,
dev.ftransz,
124.7625 + self._ftransfer_shiftz,
)
if position == 13:
umv(
dev.ftransx,
21.95 + self._ftransfer_shiftx,
dev.ftransz,
124.7750 + self._ftransfer_shiftz,
)
if position == 14:
umv(
dev.ftransx,
11.95 + self._ftransfer_shiftx,
dev.ftransz,
124.7875 + self._ftransfer_shiftz,
)
if position == 15:
umv(
dev.ftransx,
1.95 + self._ftransfer_shiftx,
dev.ftransz,
124.8000 + self._ftransfer_shiftz,
)
if position == 16:
umv(
dev.ftransx,
41.95 + self._ftransfer_shiftx,
dev.ftransz,
139.7500 + self._ftransfer_shiftz,
)
if position == 17:
umv(
dev.ftransx,
31.95 + self._ftransfer_shiftx,
dev.ftransz,
139.7625 + self._ftransfer_shiftz,
)
if position == 18:
umv(
dev.ftransx,
21.95 + self._ftransfer_shiftx,
dev.ftransz,
139.7750 + self._ftransfer_shiftz,
)
if position == 19:
umv(
dev.ftransx,
11.95 + self._ftransfer_shiftx,
dev.ftransz,
139.7875 + self._ftransfer_shiftz,
)
if position == 20:
umv(
dev.ftransx,
1.95 + self._ftransfer_shiftx,
dev.ftransz,
139.8000 + self._ftransfer_shiftz,
)
class FlomniAlignmentMixin:
default_correction_file = "correction_flomni_20210300_360deg.txt"
def reset_correction(self, use_default_correction=True):
"""
Reset the correction to the default values.
If use_default_correction is False, the correction will be set to empty values.
Otherwise the default values will be loaded.
Args:
use_default_correction (bool, optional): If set to true, a call reset the correction to the default values. Defaults to True.
"""
self.corr_pos_y = []
self.corr_angle_y = []
self.corr_pos_y_2 = []
self.corr_angle_y_2 = []
if use_default_correction:
try:
self.read_additional_correction_y(self.default_correction_file)
logger.info(f"Applying default correction from {self.default_correction_file}")
except FileNotFoundError:
logger.warning(
f"Could not find default correction file {self.default_correction_file}."
)
logger.warning("Not applying any correction.")
def reset_tomo_alignment_fit(self):
self.client.delete_global_var("tomo_alignment_fit")
def read_alignment_offset(
self,
dir_path=os.path.expanduser("~/Data10/specES1/internal/"),
setup="flomni",
use_vertical_default_values=True,
):
"""
Read the alignment offset from the given directory and set the global parameter
tomo_alignment_fit.
Args:
dir_path (str, optional): The directory to read the alignment offset from. Defaults to os.path.expanduser("~/Data10/specES1/internal/").
"""
tomo_alignment_fit = np.zeros((2, 5))
with open(os.path.join(dir_path, "ptychotomoalign_Ax.txt"), "r") as file:
tomo_alignment_fit[0][0] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Bx.txt"), "r") as file:
tomo_alignment_fit[0][1] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Cx.txt"), "r") as file:
tomo_alignment_fit[0][2] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Ay.txt"), "r") as file:
tomo_alignment_fit[1][0] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_By.txt"), "r") as file:
tomo_alignment_fit[1][1] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Cy.txt"), "r") as file:
tomo_alignment_fit[1][2] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Ay3.txt"), "r") as file:
tomo_alignment_fit[1][3] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Cy3.txt"), "r") as file:
tomo_alignment_fit[1][4] = file.readline()
print("New alignment parameters loaded:")
print(
f"X Amplitude {tomo_alignment_fit[0][0]}, "
f"X Phase {tomo_alignment_fit[0][1]}, "
f"X Offset {tomo_alignment_fit[0][2]}, "
f"Y Amplitude {tomo_alignment_fit[1][0]}, "
f"Y Phase {tomo_alignment_fit[1][1]}, "
f"Y Offset {tomo_alignment_fit[1][2]}, "
f"Y 3rd Order Amplitude {tomo_alignment_fit[1][3]}, "
f"Y 3rd Order Phase {tomo_alignment_fit[1][4]} ."
)
if use_vertical_default_values:
print(
f"Using default values for vertical alignment for setup {setup}. Optional: use_vertical_default_values=False"
)
if setup == "flomni":
tomo_alignment_fit[1][0] = 0
tomo_alignment_fit[1][1] = 0
tomo_alignment_fit[1][2] = 0
tomo_alignment_fit[1][3] = 0
tomo_alignment_fit[1][4] = 0
elif setup == "omny":
tomo_alignment_fit[1][0] = 2.588628
tomo_alignment_fit[1][1] = -2.385422
tomo_alignment_fit[1][2] = 0
tomo_alignment_fit[1][3] = 1.010583
tomo_alignment_fit[1][4] = -1.359157
print("Follwing parameters will be used:")
print(
f"X Amplitude {tomo_alignment_fit[0][0]}, "
f"X Phase {tomo_alignment_fit[0][1]}, "
f"X Offset {tomo_alignment_fit[0][2]}, "
f"Y Amplitude {tomo_alignment_fit[1][0]}, "
f"Y Phase {tomo_alignment_fit[1][1]}, "
f"Y Offset {tomo_alignment_fit[1][2]}, "
f"Y 3rd Order Amplitude {tomo_alignment_fit[1][3]}, "
f"Y 3rd Order Phase {tomo_alignment_fit[1][4]} ."
)
self.client.set_global_var("tomo_alignment_fit", tomo_alignment_fit.tolist())
# x amp, phase, offset, y amp, phase, offset, 3rd order amp, 3rd order phase
# 0 0 0 1 0 2 1 0 1 1 1 2 1 3 1 4
def get_alignment_offset(self, angle: float):
"""
Compute the alignment offset for the given angle.
Args:
angle (float): The angle to compute the alignment offset for.
Returns:
tuple: The alignment offset in x, y and z direction.
"""
tomo_alignment_fit = self.client.get_global_var("tomo_alignment_fit")
if tomo_alignment_fit is None:
print("Not applying any alignment offsets. No tomo alignment fit data available.\n")
return (0, 0, 0)
# x amp, phase, offset, y amp, phase, offset
# 0 0 0 1 0 2 1 0 1 1 1 2
correction_x = (
tomo_alignment_fit[0][0] * np.sin(np.radians(angle) + tomo_alignment_fit[0][1])
+ tomo_alignment_fit[0][2]
)
correction_y = (
tomo_alignment_fit[1][0] * np.sin(np.radians(angle) + tomo_alignment_fit[1][1])
+ tomo_alignment_fit[1][2]
+ tomo_alignment_fit[1][3] * np.sin(3 * np.radians(angle) + tomo_alignment_fit[1][4])
)
correction_z = tomo_alignment_fit[0][0] * np.sin(
np.radians(angle + 90) + tomo_alignment_fit[0][1]
)
print(
f"Alignment offset x {correction_x}, y {correction_y}, z {correction_z} for angle"
f" {angle}\n"
)
return (correction_x, correction_y, correction_z)
def _read_correction_file(self, correction_file: str):
with open(correction_file, "r") as f:
num_elements = f.readline()
int_num_elements = int(num_elements.split(" ")[2])
corr_pos = []
corr_angle = []
for j in range(int_num_elements * 2):
line = f.readline()
value = line.split(" ")[2]
name = line.split(" ")[0].split("[")[0]
if name == "corr_pos":
corr_pos.append(float(value) / 1000)
elif name == "corr_angle":
corr_angle.append(float(value))
print(
f"Loading default mirror correction from file {correction_file} containing {int_num_elements} elements."
)
return corr_pos, corr_angle
def read_additional_correction_y(self, correction_file: str):
self.corr_pos_y, self.corr_angle_y = self._read_correction_file(correction_file)
def read_additional_correction_y_2(self, correction_file: str):
self.corr_pos_y_2, self.corr_angle_y_2 = self._read_correction_file(correction_file)
def compute_additional_correction_y(self, angle):
return self._compute_additional_correction(angle, iteration=1)
def compute_additional_correction_y_2(self, angle):
return self._compute_additional_correction(angle, iteration=2)
def _compute_additional_correction(self, angle, iteration=1):
if iteration == 1:
corr_pos = self.corr_pos_y
corr_angle = self.corr_angle_y
elif iteration == 2:
corr_pos = self.corr_pos_y_2
corr_angle = self.corr_angle_y_2
if not corr_pos:
print("Not applying any additional correction. No data available.\n")
return 0
# find index of closest angle
for j, _ in enumerate(corr_pos):
newangledelta = np.fabs(corr_angle[j] - angle)
if j == 0:
angledelta = newangledelta
additional_correction_shift = corr_pos[j]
continue
if newangledelta < angledelta:
additional_correction_shift = corr_pos[j]
angledelta = newangledelta
if additional_correction_shift == 0 and angle > corr_angle[-1]:
additional_correction_shift = corr_pos[-1]
print(f"Additional correction shift {iteration} in y: {additional_correction_shift}")
return additional_correction_shift
class OMNY(
OMNYInitStagesMixin,
FlomniSampleTransferMixin,
FlomniAlignmentMixin,
OMNYSampleTransferMixin,
OMNYAlignmentMixin,
OMNYOpticsMixin,
cSAXSBeamlineChecks,
):
@@ -0,0 +1,219 @@
import time
import numpy as np
from rich import box
from rich.console import Console
from rich.table import Table
class OMNYAlignmentError(Exception):
pass
class OMNYAlignmentMixin:
default_correction_file = "correction_flomni_20210300_360deg.txt"
def reset_correction(self, use_default_correction=True):
"""
Reset the correction to the default values.
If use_default_correction is False, the correction will be set to empty values.
Otherwise the default values will be loaded.
Args:
use_default_correction (bool, optional): If set to true, a call reset the correction to the default values. Defaults to True.
"""
self.corr_pos_y = []
self.corr_angle_y = []
self.corr_pos_y_2 = []
self.corr_angle_y_2 = []
if use_default_correction:
try:
self.read_additional_correction_y(self.default_correction_file)
logger.info(f"Applying default correction from {self.default_correction_file}")
except FileNotFoundError:
logger.warning(
f"Could not find default correction file {self.default_correction_file}."
)
logger.warning("Not applying any correction.")
def reset_tomo_alignment_fit(self):
self.client.delete_global_var("tomo_alignment_fit")
def read_alignment_offset(
self,
dir_path=os.path.expanduser("~/Data10/specES1/internal/"),
setup="flomni",
use_vertical_default_values=True,
):
"""
Read the alignment offset from the given directory and set the global parameter
tomo_alignment_fit.
Args:
dir_path (str, optional): The directory to read the alignment offset from. Defaults to os.path.expanduser("~/Data10/specES1/internal/").
"""
tomo_alignment_fit = np.zeros((2, 5))
with open(os.path.join(dir_path, "ptychotomoalign_Ax.txt"), "r") as file:
tomo_alignment_fit[0][0] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Bx.txt"), "r") as file:
tomo_alignment_fit[0][1] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Cx.txt"), "r") as file:
tomo_alignment_fit[0][2] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Ay.txt"), "r") as file:
tomo_alignment_fit[1][0] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_By.txt"), "r") as file:
tomo_alignment_fit[1][1] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Cy.txt"), "r") as file:
tomo_alignment_fit[1][2] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Ay3.txt"), "r") as file:
tomo_alignment_fit[1][3] = file.readline()
with open(os.path.join(dir_path, "ptychotomoalign_Cy3.txt"), "r") as file:
tomo_alignment_fit[1][4] = file.readline()
print("New alignment parameters loaded:")
print(
f"X Amplitude {tomo_alignment_fit[0][0]}, "
f"X Phase {tomo_alignment_fit[0][1]}, "
f"X Offset {tomo_alignment_fit[0][2]}, "
f"Y Amplitude {tomo_alignment_fit[1][0]}, "
f"Y Phase {tomo_alignment_fit[1][1]}, "
f"Y Offset {tomo_alignment_fit[1][2]}, "
f"Y 3rd Order Amplitude {tomo_alignment_fit[1][3]}, "
f"Y 3rd Order Phase {tomo_alignment_fit[1][4]} ."
)
if use_vertical_default_values:
print(
f"Using default values for vertical alignment for setup {setup}. Optional: use_vertical_default_values=False"
)
if setup == "flomni":
tomo_alignment_fit[1][0] = 0
tomo_alignment_fit[1][1] = 0
tomo_alignment_fit[1][2] = 0
tomo_alignment_fit[1][3] = 0
tomo_alignment_fit[1][4] = 0
elif setup == "omny":
tomo_alignment_fit[1][0] = 2.588628
tomo_alignment_fit[1][1] = -2.385422
tomo_alignment_fit[1][2] = 0
tomo_alignment_fit[1][3] = 1.010583
tomo_alignment_fit[1][4] = -1.359157
print("Follwing parameters will be used:")
print(
f"X Amplitude {tomo_alignment_fit[0][0]}, "
f"X Phase {tomo_alignment_fit[0][1]}, "
f"X Offset {tomo_alignment_fit[0][2]}, "
f"Y Amplitude {tomo_alignment_fit[1][0]}, "
f"Y Phase {tomo_alignment_fit[1][1]}, "
f"Y Offset {tomo_alignment_fit[1][2]}, "
f"Y 3rd Order Amplitude {tomo_alignment_fit[1][3]}, "
f"Y 3rd Order Phase {tomo_alignment_fit[1][4]} ."
)
self.client.set_global_var("tomo_alignment_fit", tomo_alignment_fit.tolist())
# x amp, phase, offset, y amp, phase, offset, 3rd order amp, 3rd order phase
# 0 0 0 1 0 2 1 0 1 1 1 2 1 3 1 4
def get_alignment_offset(self, angle: float):
"""
Compute the alignment offset for the given angle.
Args:
angle (float): The angle to compute the alignment offset for.
Returns:
tuple: The alignment offset in x, y and z direction.
"""
tomo_alignment_fit = self.client.get_global_var("tomo_alignment_fit")
if tomo_alignment_fit is None:
print("Not applying any alignment offsets. No tomo alignment fit data available.\n")
return (0, 0, 0)
# x amp, phase, offset, y amp, phase, offset
# 0 0 0 1 0 2 1 0 1 1 1 2
correction_x = (
tomo_alignment_fit[0][0] * np.sin(np.radians(angle) + tomo_alignment_fit[0][1])
+ tomo_alignment_fit[0][2]
)
correction_y = (
tomo_alignment_fit[1][0] * np.sin(np.radians(angle) + tomo_alignment_fit[1][1])
+ tomo_alignment_fit[1][2]
+ tomo_alignment_fit[1][3] * np.sin(3 * np.radians(angle) + tomo_alignment_fit[1][4])
)
correction_z = tomo_alignment_fit[0][0] * np.sin(
np.radians(angle + 90) + tomo_alignment_fit[0][1]
)
print(
f"Alignment offset x {correction_x}, y {correction_y}, z {correction_z} for angle"
f" {angle}\n"
)
return (correction_x, correction_y, correction_z)
def _read_correction_file(self, correction_file: str):
with open(correction_file, "r") as f:
num_elements = f.readline()
int_num_elements = int(num_elements.split(" ")[2])
corr_pos = []
corr_angle = []
for j in range(int_num_elements * 2):
line = f.readline()
value = line.split(" ")[2]
name = line.split(" ")[0].split("[")[0]
if name == "corr_pos":
corr_pos.append(float(value) / 1000)
elif name == "corr_angle":
corr_angle.append(float(value))
print(
f"Loading default mirror correction from file {correction_file} containing {int_num_elements} elements."
)
return corr_pos, corr_angle
def read_additional_correction_y(self, correction_file: str):
self.corr_pos_y, self.corr_angle_y = self._read_correction_file(correction_file)
def read_additional_correction_y_2(self, correction_file: str):
self.corr_pos_y_2, self.corr_angle_y_2 = self._read_correction_file(correction_file)
def compute_additional_correction_y(self, angle):
return self._compute_additional_correction(angle, iteration=1)
def compute_additional_correction_y_2(self, angle):
return self._compute_additional_correction(angle, iteration=2)
def _compute_additional_correction(self, angle, iteration=1):
if iteration == 1:
corr_pos = self.corr_pos_y
corr_angle = self.corr_angle_y
elif iteration == 2:
corr_pos = self.corr_pos_y_2
corr_angle = self.corr_angle_y_2
if not corr_pos:
print("Not applying any additional correction. No data available.\n")
return 0
# find index of closest angle
for j, _ in enumerate(corr_pos):
newangledelta = np.fabs(corr_angle[j] - angle)
if j == 0:
angledelta = newangledelta
additional_correction_shift = corr_pos[j]
continue
if newangledelta < angledelta:
additional_correction_shift = corr_pos[j]
angledelta = newangledelta
if additional_correction_shift == 0 and angle > corr_angle[-1]:
additional_correction_shift = corr_pos[-1]
print(f"Additional correction shift {iteration} in y: {additional_correction_shift}")
return additional_correction_shift
@@ -0,0 +1,21 @@
import time
import numpy as np
from rich import box
from rich.console import Console
from rich.table import Table
from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen, fshclose
class OMNYToolsError(Exception):
pass
class OMNYTools:
@staticmethod
def _get_user_param_safe(device, var):
param = dev[device].user_parameter
if not param or param.get(var) is None:
raise ValueError(f"Device {device} has no user parameter definition for {var}.")
return param.get(var)
@@ -0,0 +1,55 @@
import time
import numpy as np
from rich import box
from rich.console import Console
from rich.table import Table
from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen, fshclose
class OMNYToolsError(Exception):
pass
class OMNYTools:
@staticmethod
def _get_user_param_safe(device, var):
param = dev[device].user_parameter
if not param or param.get(var) is None:
raise ValueError(f"Device {device} has no user parameter definition for {var}.")
return param.get(var)
def laser_tracker_show_all(self):
dev.rtx.controller.laser_tracker_show_all()
def laser_tracker_on(self):
dev.rtx.controller.laser_tracker_on()
time.sleep(0.2)
self._laser_tracker_check_signalstrength()
def laser_tracker_off(self):
dev.rtx.controller.laser_tracker_off()
def show_signal_strength_interferometer(self):
dev.rtx.controller.show_signal_strength_interferometer()
def rt_feedback_disable(self):
self.device_manager.devices.rtx.controller.feedback_disable()
def rt_feedback_enable_with_reset(self):
self.device_manager.devices.rtx.controller.feedback_enable_with_reset()
self.rt_feedback_status()
def rt_feedback_enable_without_reset(self):
self.device_manager.devices.rtx.controller.feedback_enable_without_reset()
self.rt_feedback_status()
def rt_feedback_status(self):
feedback_status = self.device_manager.devices.rtx.controller.feedback_is_running()
if feedback_status == True:
print("The rt feedback is \x1b[92mrunning\x1b[0m.")
else:
print("The rt feedback is \x1b[91mNOT\x1b[0m running.")
@@ -97,7 +97,7 @@ class OMNYSampleTransferMixin:
#ensure aligner is in down position
#if (!_ogalil_high_limit_set(1,7))
#{
#printf("Strange, the aligner is not in the down position. Moving down.\n")
#print("Strange, the aligner is not in the down position. Moving down.\n")
#_ogalil_drive_to_limit(1,7,1)
#}
#sleep(0.1)
@@ -722,7 +722,7 @@ class OMNYSampleTransferMixin:
# oeye_cam_in
self._omnycam_samples()
# if (_rt_status_feedback() == 0) {
# printf("May the feedback be disabled? Y/n?")
# print("May the feedback be disabled? Y/n?")
# if(!yesno(1))
# exit;
#}
@@ -835,7 +835,7 @@ class OMNYSampleTransferMixin:
raise OMNYTransferError("This routine needs commissioning after park pos of gripper and new transfer stages changed.\n")
# printf("Please confirm that the npoint piezo controller is off. y/n.")
# print("Please confirm that the npoint piezo controller is off. y/n.")
# if(!yesno(1))
# exit
@@ -876,7 +876,7 @@ class OMNYSampleTransferMixin:
raise OMNYTransferError("This routine needs commissioning after park pos of gripper and new transfer stages changed.\n")
# if(A[ofzpz]>40) {
# printf("The FZP stage is too close in z direction. Please move it to at least 60 mm or less upstream.\n")
# print("The FZP stage is too close in z direction. Please move it to at least 60 mm or less upstream.\n")
# exit
# }
@@ -898,7 +898,7 @@ class OMNYSampleTransferMixin:
# umv otransz -55.6013
# if(fabs(A[ofzpz]-77.7)>0.1)
# {
# printf("FZP did not reach target position. Potentially the motor is too hot.\n")
# print("FZP did not reach target position. Potentially the motor is too hot.\n")
# exit
# }
@@ -919,12 +919,12 @@ class OMNYSampleTransferMixin:
# #free a position
# if _ostatus==0:
# epics_put(sprintf("XOMNY-SAMPLE_DB_%s:%d.VAL", _ocontainer, _oposition),0)
# epics_put(sprintf("XOMNY-SAMPLE_DB_%s:%d.DESC", _ocontainer, _oposition),"-")
# epics_put(sprint("XOMNY-SAMPLE_DB_%s:%d.VAL", _ocontainer, _oposition),0)
# epics_put(sprint("XOMNY-SAMPLE_DB_%s:%d.DESC", _ocontainer, _oposition),"-")
# #sample at position
# if _ostatus>0:
# epics_put(sprintf("XOMNY-SAMPLE_DB_%s:%d.VAL",_ocontainer, _oposition),_ostatus)
# epics_put(sprintf("XOMNY-SAMPLE_DB_%s:%d.DESC", _ocontainer, _oposition), _oname)
# epics_put(sprint("XOMNY-SAMPLE_DB_%s:%d.VAL",_ocontainer, _oposition),_ostatus)
# epics_put(sprint("XOMNY-SAMPLE_DB_%s:%d.DESC", _ocontainer, _oposition), _oname)
# else:
# #free a position
# if _ostatus==0:
@@ -943,13 +943,13 @@ class OMNYSampleTransferMixin:
# #free a position
# if (_ocontainer=="0")
# {
# epics_put(sprintf("XOMNY-SAMPLE_DB_parking:%d.VAL", _oparkposition),0)
# epics_put(sprintf("XOMNY-SAMPLE_DB_parking:%d.DESC", _oparkposition),"-")
# epics_put(sprint("XOMNY-SAMPLE_DB_parking:%d.VAL", _oparkposition),0)
# epics_put(sprint("XOMNY-SAMPLE_DB_parking:%d.DESC", _oparkposition),"-")
# }
# else
# {
# epics_put(sprintf("XOMNY-SAMPLE_DB_parking:%d.VAL", _oparkposition),1)
# epics_put(sprintf("XOMNY-SAMPLE_DB_parking:%d.DESC", _oparkposition), _ocontainer)
# epics_put(sprint("XOMNY-SAMPLE_DB_parking:%d.VAL", _oparkposition),1)
# epics_put(sprint("XOMNY-SAMPLE_DB_parking:%d.DESC", _oparkposition), _ocontainer)
# }
# }'
@@ -993,7 +993,7 @@ class OMNYSampleTransferMixin:
# activeparkingposition = _oparkz_slot_check()
# if(activeparkingposition==10)
# exit
# installedshuttle = epics_get(sprintf("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# installedshuttle = epics_get(sprint("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# #is there a samle shuttle at this position?
# if(installedshuttle == "FZP")
# {
@@ -1002,7 +1002,7 @@ class OMNYSampleTransferMixin:
# {
# if(_picked_it==1)
# {
# samplename = epics_get(sprintf("XOMNY-SAMPLE_DB_shuttle_FZP:0.DESC"))
# samplename = epics_get(sprint("XOMNY-SAMPLE_DB_shuttle_FZP:0.DESC"))
# _otransfer_modify_sample(installedshuttle, 0, 0, "-")
# _otransfer_modify_sample("omny",110,2,samplename)
# }
@@ -1010,13 +1010,13 @@ class OMNYSampleTransferMixin:
# }
# else
# {
# printf("There is no FZP available.\n")
# print("There is no FZP available.\n")
# return(0)
# }
# }
# else
# {
# printf("There is no FZP shuttle installed at the current parkz position.\n")
# print("There is no FZP shuttle installed at the current parkz position.\n")
# return(0)
# }
# }
@@ -1027,7 +1027,7 @@ class OMNYSampleTransferMixin:
# activeparkingposition = _oparkz_slot_check()
# if(activeparkingposition==10)
# exit
# installedshuttle = epics_get(sprintf("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# installedshuttle = epics_get(sprint("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# #is there a samle shuttle at this position?
# if(installedshuttle == "OSA")
# {
@@ -1037,7 +1037,7 @@ class OMNYSampleTransferMixin:
# {
# if(_picked_it==1)
# {
# samplename = epics_get(sprintf("XOMNY-SAMPLE_DB_shuttle_OSA:%d.DESC",(_osampleposition-8)))
# samplename = epics_get(sprint("XOMNY-SAMPLE_DB_shuttle_OSA:%d.DESC",(_osampleposition-8)))
# _otransfer_modify_sample(installedshuttle, (_osampleposition-8), 0, "-")
# _otransfer_modify_sample("omny",110,3,samplename)
# }
@@ -1045,13 +1045,13 @@ class OMNYSampleTransferMixin:
# }
# else
# {
# printf("There is no OSA available.\n")
# print("There is no OSA available.\n")
# return(0)
# }
# }
# else
# {
# printf("There is no OSA shuttle installed at the current parkz position.\n")
# print("There is no OSA shuttle installed at the current parkz position.\n")
# return(0)
# }
# }
@@ -1122,7 +1122,7 @@ class OMNYSampleTransferMixin:
# { }
# else
# {
# printf("There is a sample pin in the gripper, which is not compatible with position %d\n", _osampleposition)
# print("There is a sample pin in the gripper, which is not compatible with position %d\n", _osampleposition)
# exit
# }
# #FZP holder
@@ -1131,7 +1131,7 @@ class OMNYSampleTransferMixin:
# { }
# else
# {
# printf("There is a FZP holder in the gripper, which is not compatible with position %d\n", _osampleposition)
# print("There is a FZP holder in the gripper, which is not compatible with position %d\n", _osampleposition)
# exit
# }
# #OSA holder
@@ -1140,7 +1140,7 @@ class OMNYSampleTransferMixin:
# { }
# else
# {
# printf("There is an OSA holder in the gripper, which is not compatible with position %d\n", _osampleposition)
# print("There is an OSA holder in the gripper, which is not compatible with position %d\n", _osampleposition)
# exit
# }
@@ -1173,7 +1173,7 @@ class OMNYSampleTransferMixin:
# activeparkingposition = _oparkz_slot_check()
# if(activeparkingposition==10)
# exit
# installedshuttle = epics_get(sprintf("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# installedshuttle = epics_get(sprint("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# #is there a samle shuttle at this position?
# if(installedshuttle == "FZP")
# {
@@ -1182,7 +1182,7 @@ class OMNYSampleTransferMixin:
# {
# if(_put_it==1)
# {
# samplename = epics_get(sprintf("XOMNY-SAMPLE_DB_omny:110.DESC"))
# samplename = epics_get(sprint("XOMNY-SAMPLE_DB_omny:110.DESC"))
# _otransfer_modify_sample(installedshuttle, 0, 1, samplename)
# _otransfer_modify_sample("omny",110,0,"-")
# }
@@ -1190,13 +1190,13 @@ class OMNYSampleTransferMixin:
# }
# else
# {
# printf("There is no free FZP position available.\n")
# print("There is no free FZP position available.\n")
# return(0)
# }
# }
# else
# {
# printf("There is no FZP shuttle installed at the current parkz position.\n")
# print("There is no FZP shuttle installed at the current parkz position.\n")
# return(0)
# }
# }
@@ -1208,16 +1208,16 @@ class OMNYSampleTransferMixin:
# activeparkingposition = _oparkz_slot_check()
# if(activeparkingposition==10)
# exit
# installedshuttle = epics_get(sprintf("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# installedshuttle = epics_get(sprint("XOMNY-SAMPLE_DB_parking:%d.DESC", activeparkingposition))
# #is there a samle shuttle at this position?
# if(installedshuttle == "OSA")
# {
# #is there a free OSA slot available?
# if (epics_get(sprintf("XOMNY-SAMPLE_DB_shuttle_OSA:%d.VAL",(_osampleposition-8)))==0)
# if (epics_get(sprint("XOMNY-SAMPLE_DB_shuttle_OSA:%d.VAL",(_osampleposition-8)))==0)
# {
# if(_put_it==1)
# {
# samplename = epics_get(sprintf("XOMNY-SAMPLE_DB_omny:110.DESC"))
# samplename = epics_get(sprint("XOMNY-SAMPLE_DB_omny:110.DESC"))
# _otransfer_modify_sample(installedshuttle, (_osampleposition-8), 1, samplename)
# _otransfer_modify_sample("omny",110,0,"-")
# }
@@ -1225,13 +1225,13 @@ class OMNYSampleTransferMixin:
# }
# else
# {
# printf("There is no free OSA position available.\n")
# print("There is no free OSA position available.\n")
# return(0)
# }
# }
# else
# {
# printf("There is no OSA shuttle installed at the current parkz position.\n")
# print("There is no OSA shuttle installed at the current parkz position.\n")
# return(0)
# }
# }
@@ -1265,9 +1265,9 @@ class OMNYSampleTransferMixin:
def otransfer_help(self):
printf("omny.transfer.otransfer_park_slot(slot) drive the parking station to transfer from <slot> with the gripper\n")
printf("omny.transfer.otransfer_park_loadlock_slot(slot) drive the parking station to transfer from <slot> with the loadlock\n")
printf("omny.transfer.otransfer_get_sample(position) pick with the gripper from <position>")
printf("omny.transfer.otransfer_put_sample(position) put with the gripper to <position>")
printf("To modify the storage information see dev.omny_samples.help()")
print("omny.transfer.otransfer_park_slot(slot) drive the parking station to transfer from <slot> with the gripper\n")
print("omny.transfer.otransfer_park_loadlock_slot(slot) drive the parking station to transfer from <slot> with the loadlock\n")
print("omny.transfer.otransfer_get_sample(position) pick with the gripper from <position>")
print("omny.transfer.otransfer_put_sample(position) put with the gripper to <position>")
print("To modify the storage information see dev.omny_samples.help()")