x01dcandappel_c
97ed0b8546
- Add x_ray_eye_align.py: GUI-based alignment procedure collecting sample centre at 8 angles (0-315 deg, full 360 deg rotation), submitting x/y offsets to XRayEye widget fit tab, reading sinusoidal fit parameters back via omny_xray_gui device signals - Update gui_tools.py: modernise to current BEC widget API (single new() call with object_name, _is_deleted() existence checks, add_ring() progress bar API, start/eta time display); add lamnigui_show_xeyealign_fittab() - Update alignment.py: add read_xray_eye_correction_from_gui(), remove obsolete GUI-procedure methods now in x_ray_eye_align.py, fix __init__ (remove orphaned _reset_init_values() call) - Update lamni.py: add set_client() call in __init__, add xrayeye_alignment_start(keep_shutter_open) and xrayeye_update_frame() matching FlOMNI naming convention, import XrayEyeAlignGUI - Fix LamNIMoveToScanCenter: replace arg_input/arg_bundle_size device-bundle pattern with required_kwargs; float dict keys caused msgpack >= 1.0 ValueError with strict_map_key=True - Update all lamni_move_to_scan_center() call sites to keyword arguments - Update docs/user/ptychography/lamni.md: document new commands, remove MATLAB/LabView references
313 lines
12 KiB
Python
313 lines
12 KiB
Python
import builtins
|
|
import time
|
|
|
|
import numpy as np
|
|
from bec_lib import bec_logger
|
|
from typeguard import typechecked
|
|
|
|
from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen
|
|
|
|
logger = bec_logger.logger
|
|
|
|
if builtins.__dict__.get("bec") is not None:
|
|
bec = builtins.__dict__.get("bec")
|
|
dev = builtins.__dict__.get("dev")
|
|
umv = builtins.__dict__.get("umv")
|
|
umvr = builtins.__dict__.get("umvr")
|
|
|
|
|
|
class XrayEyeAlign:
|
|
# pixel calibration, multiply to get mm
|
|
# PIXEL_CALIBRATION = 0.2/209 #.2 with binning
|
|
PIXEL_CALIBRATION = 0.2 / 218 # .2 with binning
|
|
|
|
def __init__(self, client, lamni) -> None:
|
|
self.client = client
|
|
self.lamni = lamni
|
|
self.device_manager = client.device_manager
|
|
self.scans = client.scans
|
|
self.corr_pos_x = []
|
|
self.corr_pos_y = []
|
|
self.corr_angle = []
|
|
self.corr_pos_x_2 = []
|
|
self.corr_pos_y_2 = []
|
|
self.corr_angle_2 = []
|
|
|
|
# ------------------------------------------------------------------
|
|
# Correction reset
|
|
# ------------------------------------------------------------------
|
|
|
|
def reset_correction(self):
|
|
self.corr_pos_x = []
|
|
self.corr_pos_y = []
|
|
self.corr_angle = []
|
|
|
|
def reset_correction_2(self):
|
|
self.corr_pos_x_2 = []
|
|
self.corr_pos_y_2 = []
|
|
self.corr_angle_2 = []
|
|
|
|
def reset_xray_eye_correction(self):
|
|
self.client.delete_global_var("tomo_fit_xray_eye")
|
|
|
|
# ------------------------------------------------------------------
|
|
# FOV offset properties
|
|
# ------------------------------------------------------------------
|
|
|
|
@property
|
|
def tomo_fovx_offset(self):
|
|
val = self.client.get_global_var("tomo_fov_offset")
|
|
if val is None:
|
|
return 0.0
|
|
return val[0] / 1000
|
|
|
|
@tomo_fovx_offset.setter
|
|
@typechecked
|
|
def tomo_fovx_offset(self, val: float):
|
|
val_old = self.client.get_global_var("tomo_fov_offset")
|
|
if val_old is None:
|
|
val_old = [0.0, 0.0]
|
|
self.client.set_global_var("tomo_fov_offset", [val * 1000, val_old[1]])
|
|
|
|
@property
|
|
def tomo_fovy_offset(self):
|
|
val = self.client.get_global_var("tomo_fov_offset")
|
|
if val is None:
|
|
return 0.0
|
|
return val[1] / 1000
|
|
|
|
@tomo_fovy_offset.setter
|
|
@typechecked
|
|
def tomo_fovy_offset(self, val: float):
|
|
val_old = self.client.get_global_var("tomo_fov_offset")
|
|
if val_old is None:
|
|
val_old = [0.0, 0.0]
|
|
self.client.set_global_var("tomo_fov_offset", [val_old[0], val * 1000])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
# ------------------------------------------------------------------
|
|
# X-ray eye camera control
|
|
# ------------------------------------------------------------------
|
|
|
|
def update_fov(self, k: int):
|
|
self._xray_fov_xy[0] = max(epics_get(f"XOMNYI-XEYE-XWIDTH_X:{k}"), self._xray_fov_xy[0])
|
|
self._xray_fov_xy[1] = max(0, self._xray_fov_xy[0])
|
|
|
|
# ------------------------------------------------------------------
|
|
# Alignment procedure
|
|
# ------------------------------------------------------------------
|
|
|
|
# ------------------------------------------------------------------
|
|
# Alignment output
|
|
# ------------------------------------------------------------------
|
|
|
|
# ------------------------------------------------------------------
|
|
# X-ray eye sinusoidal correction (loaded from MATLAB fit files)
|
|
# ------------------------------------------------------------------
|
|
|
|
def read_xray_eye_correction(self, dir_path=None):
|
|
import os
|
|
if dir_path is None:
|
|
dir_path = os.path.expanduser("~/Data10/specES1/internal/")
|
|
tomo_fit_xray_eye = np.zeros((2, 3))
|
|
for i, axis in enumerate(["x", "y"]):
|
|
for j, coeff in enumerate(["A", "B", "C"]):
|
|
with open(os.path.join(dir_path, f"ptychotomoalign_{coeff}{axis}.txt"), "r") as f:
|
|
tomo_fit_xray_eye[i][j] = f.readline()
|
|
|
|
self.client.set_global_var("tomo_fit_xray_eye", tomo_fit_xray_eye.tolist())
|
|
# x amp, phase, offset, y amp, phase, offset
|
|
# 0 0 0 1 0 2 1 0 1 1 1 2
|
|
print("New alignment parameters loaded from X-ray eye")
|
|
|
|
def read_xray_eye_correction_from_gui(self):
|
|
"""Read the sinusoidal fit from the XRayEye GUI widget.
|
|
|
|
Replaces the MATLAB file read (read_xray_eye_correction) when the
|
|
BEC GUI alignment has been performed. Reads fit_params_x and
|
|
fit_params_y from the omny_xray_gui device and stores the result as
|
|
tomo_fit_xray_eye = [[Ax, Bx, Cx], [Ay, By, Cy]] in the BEC global
|
|
variable store.
|
|
|
|
The stored array is consumed by lamni_compute_additional_correction_xeye_mu().
|
|
|
|
Important: this method reads from the live GUI widget via the device
|
|
(omny_xray_gui). If the XRayEye GUI window has been closed since the
|
|
alignment was performed, the fit parameters will no longer be available
|
|
and this call will fail. In that case use read_xray_eye_correction()
|
|
to reload from the archived text files written by write_output().
|
|
"""
|
|
import builtins
|
|
dev = builtins.__dict__.get("dev")
|
|
|
|
tomo_fit_xray_eye = np.zeros((2, 3))
|
|
|
|
params_x = dev.omny_xray_gui.fit_params_x.get()
|
|
tomo_fit_xray_eye[0][0] = params_x["SineModel_0_amplitude"]
|
|
tomo_fit_xray_eye[0][1] = params_x["SineModel_0_shift"]
|
|
tomo_fit_xray_eye[0][2] = params_x["LinearModel_1_intercept"]
|
|
|
|
params_y = dev.omny_xray_gui.fit_params_y.get()
|
|
tomo_fit_xray_eye[1][0] = params_y["SineModel_0_amplitude"]
|
|
tomo_fit_xray_eye[1][1] = params_y["SineModel_0_shift"]
|
|
tomo_fit_xray_eye[1][2] = params_y["LinearModel_1_intercept"]
|
|
|
|
self.client.set_global_var("tomo_fit_xray_eye", tomo_fit_xray_eye.tolist())
|
|
print("New alignment parameters loaded from XRayEye GUI fit:")
|
|
print(
|
|
f" X: A={tomo_fit_xray_eye[0][0]:.4f}, "
|
|
f"B={tomo_fit_xray_eye[0][1]:.4f}, "
|
|
f"C={tomo_fit_xray_eye[0][2]:.4f}"
|
|
)
|
|
print(
|
|
f" Y: A={tomo_fit_xray_eye[1][0]:.4f}, "
|
|
f"B={tomo_fit_xray_eye[1][1]:.4f}, "
|
|
f"C={tomo_fit_xray_eye[1][2]:.4f}"
|
|
)
|
|
|
|
|
|
def read_xray_eye_correction_from_gui(self):
|
|
"""Read the sinusoidal fit from the XRayEye GUI widget.
|
|
|
|
Replaces the MATLAB file read (read_xray_eye_correction) when the
|
|
BEC GUI alignment has been performed. Reads fit_params_x and
|
|
fit_params_y from the omny_xray_gui device and stores the result as
|
|
tomo_fit_xray_eye = [[Ax, Bx, Cx], [Ay, By, Cy]] in the BEC global
|
|
variable store.
|
|
|
|
The stored array is consumed by lamni_compute_additional_correction_xeye_mu().
|
|
"""
|
|
import builtins
|
|
dev = builtins.__dict__.get("dev")
|
|
|
|
tomo_fit_xray_eye = np.zeros((2, 3))
|
|
|
|
params_x = dev.omny_xray_gui.fit_params_x.get()
|
|
tomo_fit_xray_eye[0][0] = params_x["SineModel_0_amplitude"]
|
|
tomo_fit_xray_eye[0][1] = params_x["SineModel_0_shift"]
|
|
tomo_fit_xray_eye[0][2] = params_x["LinearModel_1_intercept"]
|
|
|
|
params_y = dev.omny_xray_gui.fit_params_y.get()
|
|
tomo_fit_xray_eye[1][0] = params_y["SineModel_0_amplitude"]
|
|
tomo_fit_xray_eye[1][1] = params_y["SineModel_0_shift"]
|
|
tomo_fit_xray_eye[1][2] = params_y["LinearModel_1_intercept"]
|
|
|
|
self.client.set_global_var("tomo_fit_xray_eye", tomo_fit_xray_eye.tolist())
|
|
print("New alignment parameters loaded from XRayEye GUI fit:")
|
|
print(
|
|
f" X: A={tomo_fit_xray_eye[0][0]:.4f}, "
|
|
f"B={tomo_fit_xray_eye[0][1]:.4f}, "
|
|
f"C={tomo_fit_xray_eye[0][2]:.4f}"
|
|
)
|
|
print(
|
|
f" Y: A={tomo_fit_xray_eye[1][0]:.4f}, "
|
|
f"B={tomo_fit_xray_eye[1][1]:.4f}, "
|
|
f"C={tomo_fit_xray_eye[1][2]:.4f}"
|
|
)
|
|
|
|
print(
|
|
f"X Amplitude {tomo_fit_xray_eye[0][0]}, "
|
|
f"X Phase {tomo_fit_xray_eye[0][1]}, "
|
|
f"X Offset {tomo_fit_xray_eye[0][2]}, "
|
|
f"Y Amplitude {tomo_fit_xray_eye[1][0]}, "
|
|
f"Y Phase {tomo_fit_xray_eye[1][1]}, "
|
|
f"Y Offset {tomo_fit_xray_eye[1][2]}"
|
|
)
|
|
|
|
def lamni_compute_additional_correction_xeye_mu(self, angle):
|
|
"""Compute sinusoidal correction from the X-ray eye fit for the given angle."""
|
|
tomo_fit_xray_eye = self.client.get_global_var("tomo_fit_xray_eye")
|
|
if tomo_fit_xray_eye is None:
|
|
print("Not applying any additional correction. No x-ray eye data available.\n")
|
|
return (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_fit_xray_eye[0][0] * np.sin(np.radians(angle) + tomo_fit_xray_eye[0][1])
|
|
+ tomo_fit_xray_eye[0][2]
|
|
) / 1000
|
|
correction_y = (
|
|
tomo_fit_xray_eye[1][0] * np.sin(np.radians(angle) + tomo_fit_xray_eye[1][1])
|
|
+ tomo_fit_xray_eye[1][2]
|
|
) / 1000
|
|
|
|
print(f"Xeye correction x {correction_x}, y {correction_y} for angle {angle}\n")
|
|
return (correction_x, correction_y)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Additional lookup-table corrections (iteration 1 and 2)
|
|
# ------------------------------------------------------------------
|
|
|
|
def read_additional_correction(self, correction_file: str):
|
|
self.corr_pos_x, self.corr_pos_y, self.corr_angle = self._read_correction_file_xy(
|
|
correction_file
|
|
)
|
|
|
|
def read_additional_correction_2(self, correction_file: str):
|
|
self.corr_pos_x_2, self.corr_pos_y_2, self.corr_angle_2 = self._read_correction_file_xy(
|
|
correction_file
|
|
)
|
|
|
|
def _read_correction_file_xy(self, correction_file: str):
|
|
"""Parse a correction file that contains corr_pos_x, corr_pos_y and corr_angle entries."""
|
|
with open(correction_file, "r") as f:
|
|
num_elements = f.readline()
|
|
int_num_elements = int(num_elements.split(" ")[2])
|
|
print(int_num_elements)
|
|
corr_pos_x = []
|
|
corr_pos_y = []
|
|
corr_angle = []
|
|
for j in range(0, int_num_elements * 3):
|
|
line = f.readline()
|
|
value = line.split(" ")[2]
|
|
name = line.split(" ")[0].split("[")[0]
|
|
if name == "corr_pos_x":
|
|
corr_pos_x.append(float(value) / 1000)
|
|
elif name == "corr_pos_y":
|
|
corr_pos_y.append(float(value) / 1000)
|
|
elif name == "corr_angle":
|
|
corr_angle.append(float(value))
|
|
return corr_pos_x, corr_pos_y, corr_angle
|
|
|
|
def compute_additional_correction(self, angle):
|
|
return self._compute_correction_xy(
|
|
angle, self.corr_pos_x, self.corr_pos_y, self.corr_angle, label="1"
|
|
)
|
|
|
|
def compute_additional_correction_2(self, angle):
|
|
return self._compute_correction_xy(
|
|
angle, self.corr_pos_x_2, self.corr_pos_y_2, self.corr_angle_2, label="2"
|
|
)
|
|
|
|
def _compute_correction_xy(self, angle, corr_pos_x, corr_pos_y, corr_angle, label=""):
|
|
"""Find the correction for the closest angle in the lookup table."""
|
|
if not corr_pos_x:
|
|
print(f"Not applying additional correction {label}. No data available.\n")
|
|
return (0, 0)
|
|
|
|
shift_x = corr_pos_x[0]
|
|
shift_y = corr_pos_y[0]
|
|
angledelta = np.fabs(corr_angle[0] - angle)
|
|
|
|
for j in range(1, len(corr_pos_x)):
|
|
newangledelta = np.fabs(corr_angle[j] - angle)
|
|
if newangledelta < angledelta:
|
|
shift_x = corr_pos_x[j]
|
|
shift_y = corr_pos_y[j]
|
|
angledelta = newangledelta
|
|
|
|
if shift_x == 0 and angle < corr_angle[0]:
|
|
shift_x = corr_pos_x[0]
|
|
shift_y = corr_pos_y[0]
|
|
|
|
if shift_x == 0 and angle > corr_angle[-1]:
|
|
shift_x = corr_pos_x[-1]
|
|
shift_y = corr_pos_y[-1]
|
|
|
|
print(f"Additional correction shifts {label}: {shift_x} {shift_y}")
|
|
return (shift_x, shift_y)
|