bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Feat/update lamni to flomni standards run1 #258

Merged
holler merged 16 commits from feat/update_lamni_to_flomni_standards into main 2026-07-12 19:07:18 +02:00
Conversation 0 Commits 16 Files Changed 23
+2112 -940
23 changed files with 2112 additions and 940 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/LamNI/__init__.py
+8 -3
View File
@@ -1,6 +1,11 @@
from .alignment import XrayEyeAlign
from .lamni_alignment_mixin import LamNIAlignmentMixin
from .lamni import LamNI
from .lamni_optics_mixin import LamNIInitError, LaMNIInitStages, LamNIOpticsMixin
__all__ = [
"LamNI", "XrayEyeAlign", "LamNIInitError", "LaMNIInitStages", "LamNIOpticsMixin"
]
"LamNI",
"LamNIAlignmentMixin",
"LamNIInitError",
"LaMNIInitStages",
"LamNIOpticsMixin",
]
csaxs_bec/bec_ipython_client/plugins/LamNI/alignment.py
-461
View File
@@ -1,461 +0,0 @@
import builtins
import time
from collections import defaultdict
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.alignment_values = defaultdict(list)
self._reset_init_values()
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
# ------------------------------------------------------------------
def _reset_init_values(self):
self.shift_xy = [0, 0]
self._xray_fov_xy = [0, 0]
def _disable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_disable()
def _enable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_enable_with_reset()
def tomo_rotate(self, val: float):
# pylint: disable=undefined-variable
umv(self.device_manager.devices.lsamrot, val)
def get_tomo_angle(self):
return self.device_manager.devices.lsamrot.readback.read()["lsamrot"]["value"]
# ------------------------------------------------------------------
# X-ray eye camera control
# ------------------------------------------------------------------
def save_frame(self):
epics_put("XOMNYI-XEYE-SAVFRAME:0", 1)
def update_frame(self):
epics_put("XOMNYI-XEYE-ACQDONE:0", 0)
# start live
epics_put("XOMNYI-XEYE-ACQ:0", 1)
# wait for start live
while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0:
time.sleep(0.5)
print("waiting for live view to start...")
fshopen()
epics_put("XOMNYI-XEYE-ACQDONE:0", 0)
while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0:
print("waiting for new frame...")
time.sleep(0.5)
time.sleep(0.5)
# stop live view
epics_put("XOMNYI-XEYE-ACQ:0", 0)
time.sleep(1)
print("got new frame")
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])
@property
def movement_buttons_enabled(self):
return [epics_get("XOMNYI-XEYE-ENAMVX:0"), epics_get("XOMNYI-XEYE-ENAMVY:0")]
@movement_buttons_enabled.setter
def movement_buttons_enabled(self, enabled: bool):
enabled = int(enabled)
epics_put("XOMNYI-XEYE-ENAMVX:0", enabled)
epics_put("XOMNYI-XEYE-ENAMVY:0", enabled)
def send_message(self, msg: str):
epics_put("XOMNYI-XEYE-MESSAGE:0.DESC", msg)
# ------------------------------------------------------------------
# Alignment procedure
# ------------------------------------------------------------------
def align(self):
self._reset_init_values()
self.reset_correction()
self.reset_correction_2()
self._disable_rt_feedback()
epics_put("XOMNYI-XEYE-PIXELSIZE:0", self.PIXEL_CALIBRATION)
self._enable_rt_feedback()
self.movement_buttons_enabled = False
epics_put("XOMNYI-XEYE-ACQ:0", 0)
self.send_message("please wait...")
epics_put("XOMNYI-XEYE-SAMPLENAME:0.DESC", "Let us LAMNI...")
self._disable_rt_feedback()
k = 0
self.lamni.lfzp_in()
self.update_frame()
self.movement_buttons_enabled = False
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
epics_put("XOMNYI-XEYE-STEP:0", 0)
self.send_message("Submit center value of FZP.")
while True:
if epics_get("XOMNYI-XEYE-SUBMIT:0") == 1:
val_x = epics_get(f"XOMNYI-XEYE-XVAL_X:{k}") * self.PIXEL_CALIBRATION # in mm
val_y = epics_get(f"XOMNYI-XEYE-YVAL_Y:{k}") * self.PIXEL_CALIBRATION # in mm
self.alignment_values[k] = [val_x, val_y]
print(
f"Clicked position {k}: x {self.alignment_values[k][0]}, y"
f" {self.alignment_values[k][1]}"
)
if k == 0: # received center value of FZP
self.send_message("please wait ...")
self.lamni.loptics_out()
epics_put("XOMNYI-XEYE-SUBMIT:0", -1)
self.movement_buttons_enabled = False
print("Moving sample in, FZP out")
self._disable_rt_feedback()
time.sleep(0.3)
self._enable_rt_feedback()
time.sleep(0.3)
self.update_frame()
self.send_message("Go and find the sample")
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
self.movement_buttons_enabled = True
elif k == 1: # received sample center value at samrot 0
msg = (
f"Base shift values from movement are x {self.shift_xy[0]}, y"
f" {self.shift_xy[1]}"
)
print(msg)
logger.info(msg)
self.shift_xy[0] += (
self.alignment_values[0][0] - self.alignment_values[1][0]
) * 1000
self.shift_xy[1] += (
self.alignment_values[1][1] - self.alignment_values[0][1]
) * 1000
print(
"Base shift values from movement and clicked position are x"
f" {self.shift_xy[0]}, y {self.shift_xy[1]}"
)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000, self.shift_xy[1] / 1000, self.get_tomo_angle()
).wait()
self.send_message("please wait ...")
epics_put("XOMNYI-XEYE-SUBMIT:0", -1)
self.movement_buttons_enabled = False
time.sleep(1)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000, self.shift_xy[1] / 1000, self.get_tomo_angle()
).wait()
epics_put("XOMNYI-XEYE-ANGLE:0", self.get_tomo_angle())
self.update_frame()
self.send_message("Submit sample center and FOV (0 deg)")
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
self.update_fov(k)
elif 1 < k < 10: # received sample center value at samrot 0 ... 315
self.send_message("please wait ...")
epics_put("XOMNYI-XEYE-SUBMIT:0", -1)
self._disable_rt_feedback()
self.tomo_rotate((k - 1) * 45 - 45 / 2)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000, self.shift_xy[1] / 1000, self.get_tomo_angle()
).wait()
self._disable_rt_feedback()
self.tomo_rotate((k - 1) * 45)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000, self.shift_xy[1] / 1000, self.get_tomo_angle()
).wait()
epics_put("XOMNYI-XEYE-ANGLE:0", self.get_tomo_angle())
self.update_frame()
self.send_message("Submit sample center")
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
epics_put("XOMNYI-XEYE-ENAMVX:0", 1)
self.update_fov(k)
elif k == 10: # received sample center value at samrot 270, done
self.send_message("done...")
epics_put("XOMNYI-XEYE-SUBMIT:0", -1)
self.movement_buttons_enabled = False
self.update_fov(k)
break
k += 1
epics_put("XOMNYI-XEYE-STEP:0", k)
if k < 2:
_xrayeyalignmvx = epics_get("XOMNYI-XEYE-MVX:0")
_xrayeyalignmvy = epics_get("XOMNYI-XEYE-MVY:0")
if _xrayeyalignmvx != 0 or _xrayeyalignmvy != 0:
self.shift_xy[0] = self.shift_xy[0] + _xrayeyalignmvx
self.shift_xy[1] = self.shift_xy[1] + _xrayeyalignmvy
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000, self.shift_xy[1] / 1000, self.get_tomo_angle()
).wait()
print(
f"Current center horizontal {self.shift_xy[0]} vertical {self.shift_xy[1]}"
)
epics_put("XOMNYI-XEYE-MVY:0", 0)
epics_put("XOMNYI-XEYE-MVX:0", 0)
self.update_frame()
time.sleep(0.2)
self.write_output()
fovx = self._xray_fov_xy[0] * self.PIXEL_CALIBRATION * 1000 / 2
fovy = self._xray_fov_xy[1] * self.PIXEL_CALIBRATION * 1000 / 2
print(
f"The largest field of view from the xrayeyealign was \nfovx = {fovx:.0f} microns,"
f" fovy = {fovy:.0f} microns"
)
print("Use matlab routine to fit the current alignment...")
print(
"This additional shift is applied to the base shift values\n which are x"
f" {self.shift_xy[0]}, y {self.shift_xy[1]}"
)
self._disable_rt_feedback()
self.tomo_rotate(0)
print(
"\n\nNEXT LOAD ALIGNMENT PARAMETERS\nby running"
" lamni.align.read_xray_eye_correction()\n"
)
self.client.set_global_var("tomo_fov_offset", self.shift_xy)
# ------------------------------------------------------------------
# Alignment output
# ------------------------------------------------------------------
def write_output(self):
import os
with open(
os.path.expanduser("~/Data10/specES1/internal/xrayeye_alignmentvalues"), "w"
) as alignment_values_file:
alignment_values_file.write("angle\thorizontal\tvertical\n")
for k in range(2, 11):
fovx_offset = (self.alignment_values[0][0] - self.alignment_values[k][0]) * 1000
fovy_offset = (self.alignment_values[k][1] - self.alignment_values[0][1]) * 1000
print(
f"Writing to file new alignment: number {k}, value x {fovx_offset}, y"
f" {fovy_offset}"
)
alignment_values_file.write(f"{(k-2)*45}\t{fovx_offset}\t{fovy_offset}\n")
# ------------------------------------------------------------------
# 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")
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)
csaxs_bec/bec_ipython_client/plugins/flomni/docs/2022_05_22_Aberrated 30 nm zoneplates.pdf → csaxs_bec/bec_ipython_client/plugins/LamNI/docs/2022_05_22_Aberrated 30 nm zoneplates.pdf
View File
csaxs_bec/bec_ipython_client/plugins/LamNI/extra_tomo.py
+1 -1
View File
@@ -47,7 +47,7 @@ class MagLamNI(LamNI):
for target_angle in np.linspace(current_angle, angle, steps, endpoint=True):
umv(dev.lsamrot, target_angle)
scans.lamni_move_to_scan_center(
self.align.tomo_fovx_offset, self.align.tomo_fovy_offset, target_angle
self.tomo_fovx_offset, self.tomo_fovy_offset, target_angle
)
def _at_each_angle(self, angle: float) -> None:
csaxs_bec/bec_ipython_client/plugins/LamNI/gui_tools.py
+167 -76
View File
@@ -1,14 +1,15 @@
from __future__ import annotations
import builtins
from bec_widgets.cli.client import BECDockArea
# from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen, fshclose
import datetime
import time
if builtins.__dict__.get("bec") is not None:
bec = builtins.__dict__.get("bec")
dev = builtins.__dict__.get("dev")
scans = builtins.__dict__.get("scans")
def umv(*args):
return scans.umv(*args, relative=False)
@@ -18,53 +19,102 @@ class LamniGuiToolsError(Exception):
class LamniGuiTools:
GUI_RPC_TIMEOUT = 20
def __init__(self):
self.lamni_window = None
self.text_box = None
self.progressbar = None
self.xeyegui = None
self.pdf_viewer = None
self.idle_text_box = None
def set_client(self, client):
self.client = client
self.gui = self.client.gui
self.gui = getattr(client, "gui", None)
# ------------------------------------------------------------------
# Window management
# ------------------------------------------------------------------
def lamnigui_show_gui(self):
if "lamni" in self.gui.windows:
self.gui.lamni.show()
self.lamni_window = self.gui.windows["lamni"]
self.gui.lamni.raise_window()
else:
self.gui.new("lamni")
def lamnigui_stop_gui(self):
self.gui.lamni.hide()
self.lamni_window = self.gui.new("lamni", timeout=self.GUI_RPC_TIMEOUT)
time.sleep(1)
def lamnigui_raise(self):
self.gui.lamni.raise_window()
def lamnigui_show_xeyealign(self):
self.lamnigui_show_gui()
if self._lamnigui_check_attribute_not_exists("xeyegui"):
self.lamnigui_remove_all_docks()
self.xeyegui = self.gui.lamni.new("xeyegui").new("XRayEye")
# start live
if not dev.cam_xeye.live_mode:
dev.cam_xeye.live_mode = True
def lamnigui_stop_gui(self):
self.gui.lamni.hide()
# ------------------------------------------------------------------
# Widget existence checks
# ------------------------------------------------------------------
def _lamnigui_check_attribute_not_exists(self, attribute_name):
if hasattr(self.gui,"lamni"):
if hasattr(self.gui.lamni,attribute_name):
return False
return True
def _lamnigui_is_missing(self, attribute_name: str) -> bool:
"""Check whether a stored widget reference is absent or has been deleted."""
widget = getattr(self, attribute_name, None)
if widget is None:
return True
if hasattr(widget, "_is_deleted") and widget._is_deleted():
return True
return False
# ------------------------------------------------------------------
# Dock management
# ------------------------------------------------------------------
def lamnigui_remove_all_docks(self):
self.gui.lamni.delete_all()
if hasattr(self.gui, "lamni"):
self.gui.lamni.delete_all(timeout=self.GUI_RPC_TIMEOUT)
self.progressbar = None
self.text_box = None
self.xeyegui = None
self.pdf_viewer = None
self.idle_text_box = None
# ------------------------------------------------------------------
# X-ray eye alignment views
# ------------------------------------------------------------------
def lamnigui_show_xeyealign(self):
"""Open (or raise) the X-ray eye widget on the Alignment tab."""
self.lamnigui_show_gui()
if self._lamnigui_is_missing("xeyegui"):
self.lamnigui_remove_all_docks()
self.xeyegui = self.gui.lamni.new(
"XRayEye", object_name="xrayeye", timeout=self.GUI_RPC_TIMEOUT
)
# start live
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
self.xeyegui.switch_tab("alignment")
def lamnigui_show_xeyealign_fittab(self):
"""Open (or raise) the X-ray eye widget on the Fit tab."""
self.lamnigui_show_gui()
if self._lamnigui_is_missing("xeyegui"):
self.lamnigui_remove_all_docks()
self.xeyegui = self.gui.lamni.new(
"XRayEye", object_name="xrayeye", timeout=self.GUI_RPC_TIMEOUT
)
self.xeyegui.switch_tab("fit")
# ------------------------------------------------------------------
# Idle splash
# ------------------------------------------------------------------
def lamnigui_idle(self):
self.lamnigui_show_gui()
if self._lamnigui_check_attribute_not_exists("idle_text_box"):
if self._lamnigui_is_missing("idle_text_box"):
self.lamnigui_remove_all_docks()
idle_text_box = self.gui.lamni.new("idle_textbox").new("TextBox")
self.idle_text_box = self.gui.lamni.new(
"TextBox", object_name="idle_textbox", timeout=self.GUI_RPC_TIMEOUT
)
text = (
"<pre>"
+ "██████╗ ███████╗ ██████╗ ██╗ █████╗ ███╗ ███╗███╗ ██╗██╗\n"
@@ -75,18 +125,24 @@ class LamniGuiTools:
+ "╚═════╝ ╚══════╝ ╚═════╝ ╚══════╝╚═╝ ╚═╝╚═╝ ╚═╝╚═╝ ╚═══╝╚═╝\n"
+ "</pre>"
)
idle_text_box.set_html_text(text)
self.idle_text_box.set_html_text(text)
# ------------------------------------------------------------------
# Documentation viewer
# ------------------------------------------------------------------
def lamnigui_docs(self, filename: str | None = None):
import csaxs_bec
from pathlib import Path
print("The general lamni documentation is at \nhttps://sls-csaxs.readthedocs.io/en/latest/user/ptychography/lamni.html#user-ptychography-lamni")
print(
"The general LamNI documentation is at \n"
"https://sls-csaxs.readthedocs.io/en/latest/user/ptychography/lamni.html"
)
csaxs_bec_basepath = Path(csaxs_bec.__file__).parent
docs_folder = (
csaxs_bec_basepath /
"bec_ipython_client" / "plugins" / "lamni" / "docs"
csaxs_bec_basepath / "bec_ipython_client" / "plugins" / "LamNI" / "docs"
)
if not docs_folder.is_dir():
@@ -96,20 +152,19 @@ class LamniGuiTools:
if not pdfs:
raise FileNotFoundError(f"No PDF files found in {docs_folder}")
# --- Resolve PDF ------------------------------------------------------
if filename is not None:
pdf_file = docs_folder / filename
if not pdf_file.exists():
raise FileNotFoundError(f"Requested file not found: {filename}")
else:
print("\nAvailable lamni documentation PDFs:\n")
print("\nAvailable LamNI documentation PDFs:\n")
for i, pdf in enumerate(pdfs, start=1):
print(f" {i:2d}) {pdf.name}")
print()
while True:
try:
choice = int(input(f"Select a file (1–{len(pdfs)}): "))
choice = int(input(f"Select a file (1-{len(pdfs)}): "))
if 1 <= choice <= len(pdfs):
pdf_file = pdfs[choice - 1]
break
@@ -117,62 +172,97 @@ class LamniGuiTools:
except ValueError:
print("Invalid input. Please enter a number.")
# --- GUI handling (active existence check) ----------------------------
self.lamnigui_show_gui()
if self._lamnigui_check_attribute_not_exists("PdfViewerWidget"):
if self._lamnigui_is_missing("pdf_viewer"):
self.lamnigui_remove_all_docks()
self.pdf_viewer = self.gui.lamni.new(widget="PdfViewerWidget")
self.pdf_viewer = self.gui.lamni.new(
widget="PdfViewerWidget", timeout=self.GUI_RPC_TIMEOUT
)
# --- Load PDF ---------------------------------------------------------
self.pdf_viewer.PdfViewerWidget.load_pdf(str(pdf_file.resolve()))
self.pdf_viewer.load_pdf(str(pdf_file.resolve()))
print(f"\nLoaded: {pdf_file.name}\n")
def _lamnicam_check_device_exists(self, device):
try:
device
except:
return False
else:
return True
# ------------------------------------------------------------------
# Progress bar
# ------------------------------------------------------------------
def lamnigui_show_progress(self):
self.lamnigui_show_gui()
if self._lamnigui_check_attribute_not_exists("progressbar"):
if self._lamnigui_is_missing("progressbar"):
self.lamnigui_remove_all_docks()
# Add a new dock with a RingProgressBar widget
self.progressbar = self.gui.lamni.new("progressbar").new("RingProgressBar")
# Customize the size of the progress ring
self.progressbar.set_line_widths(20)
# Disable automatic updates and manually set the self.progressbar value
self.progressbar.enable_auto_updates(False)
# Set precision for the self.progressbar display
self.progressbar.set_precision(1) # Display self.progressbar with one decimal places
# Setting multiple rigns with different values
self.progressbar.set_number_of_bars(3)
self.progressbar.rings[0].set_update("manual")
self.progressbar.rings[1].set_update("manual")
self.progressbar.rings[2].set_update("scan")
# Set the values of the rings to 50, 75, and 25 from outer to inner ring
# self.progressbar.set_value([50, 75])
# Add a new dock with a TextBox widget
self.text_box = self.gui.lamni.new(name="progress_text").new("TextBox")
self.progressbar = self.gui.lamni.new(
"RingProgressBar", timeout=self.GUI_RPC_TIMEOUT
)
# Outer ring: overall tomo progress (manual update)
self.progressbar.add_ring().set_update("manual")
# Middle ring: current sub-tomo progress (manual update)
self.progressbar.add_ring().set_update("manual")
# Inner ring: current scan progress (driven by BEC scan events)
self.progressbar.add_ring().set_update("scan")
self._lamnigui_update_progress()
def _lamnigui_update_progress(self):
if self.progressbar is not None:
progress = self.progress["projection"] / self.progress["total_projections"] * 100
subtomo_progress = (
self.progress["subtomo_projection"]
/ self.progress["subtomo_total_projections"]
* 100
"""Update the progress ring bar and centre label from the current progress state.
``self.progress`` is backed by the BEC global variable ``tomo_progress``
so this reflects the live state accessible from any BEC client session via::
client.get_global_var("tomo_progress")
"""
if self.progressbar is None:
return
main_progress_ring = self.progressbar.rings[0]
subtomo_progress_ring = self.progressbar.rings[1]
progress = self.progress["projection"] / self.progress["total_projections"] * 100
subtomo_progress = (
self.progress["subtomo_projection"]
/ self.progress["subtomo_total_projections"]
* 100
)
main_progress_ring.set_value(progress)
subtomo_progress_ring.set_value(subtomo_progress)
# Format start time
start_str = self.progress.get("tomo_start_time")
if start_str is not None:
start_display = datetime.datetime.fromisoformat(start_str).strftime(
"%Y-%m-%d %H:%M:%S"
)
self.progressbar.set_value([progress, subtomo_progress, 0])
if self.text_box is not None:
text = f"Progress report:\n Tomo type: ....................... {self.progress['tomo_type']}\n Projection: ...................... {self.progress['projection']:.0f}\n Total projections expected ....... {self.progress['total_projections']}\n Angle: ........................... {self.progress['angle']}\n Current subtomo: ................. {self.progress['subtomo']}\n Current projection within subtomo: {self.progress['subtomo_projection']}\n Total projections per subtomo: ... {self.progress['subtomo_total_projections']}"
self.text_box.set_plain_text(text)
else:
start_display = "N/A"
# Format estimated remaining time
remaining_s = self.progress.get("estimated_remaining_time")
if remaining_s is not None and remaining_s >= 0:
remaining_s = int(remaining_s)
h, rem = divmod(remaining_s, 3600)
m, s = divmod(rem, 60)
if h > 0:
eta_display = f"{h}h {m:02d}m {s:02d}s"
elif m > 0:
eta_display = f"{m}m {s:02d}s"
else:
eta_display = f"{s}s"
else:
eta_display = "N/A"
text = (
f"Progress report:\n"
f" Tomo type: {self.progress['tomo_type']}\n"
f" Projection: {self.progress['projection']:.0f}\n"
f" Total projections expected: {self.progress['total_projections']:.1f}\n"
f" Angle: {self.progress['angle']:.1f}\n"
f" Current subtomo: {self.progress['subtomo']}\n"
f" Current projection within subtomo: {self.progress['subtomo_projection']}\n"
f" Total projections per subtomo: {int(self.progress['subtomo_total_projections'])}\n"
f" Scan started: {start_display}\n"
f" Est. remaining: {eta_display}"
)
self.progressbar.set_center_label(text)
if __name__ == "__main__":
@@ -183,6 +273,7 @@ if __name__ == "__main__":
client.start()
client.gui = BECGuiClient()
lamni_gui = LamniGuiTools(client)
lamni_gui = LamniGuiTools()
lamni_gui.set_client(client)
lamni_gui.lamnigui_show_gui()
lamni_gui.lamnigui_show_progress()
lamni_gui.lamnigui_show_progress()
csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py
+337 -47
View File
@@ -17,8 +17,9 @@ from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import (
TomoIDManager,
)
from csaxs_bec.bec_ipython_client.plugins.LamNI.gui_tools import LamniGuiTools
from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_alignment_mixin import LamNIAlignmentMixin
from .alignment import XrayEyeAlign
from .x_ray_eye_align import XrayEyeAlign as XrayEyeAlignGUI
from .lamni_optics_mixin import LaMNIInitStages, LamNIOpticsMixin
logger = bec_logger.logger
@@ -31,14 +32,94 @@ if builtins.__dict__.get("bec") is not None:
umvr = builtins.__dict__.get("umvr")
class LamNI(LamNIOpticsMixin, LamniGuiTools):
class _ProgressProxy:
"""Dict-like proxy that persists the LamNI progress dict as a BEC global variable.
Every read (`proxy["key"]`) fetches the current dict from the global var store,
and every write (`proxy["key"] = val`) fetches, updates, and saves it back.
This makes the progress state visible to all BEC client sessions via
``client.get_global_var("tomo_progress")`` -- same key and shape as the
FlOMNI side, since only one tomography setup is ever live against a given
BEC session.
"""
_GLOBAL_VAR_KEY = "tomo_progress"
_DEFAULTS: dict = {
"tomo_type": "Equally spaced sub-tomograms",
"subtomo": 0,
"subtomo_projection": 0,
"subtomo_total_projections": 1,
"projection": 0,
"total_projections": 1,
"angle": 0,
"tomo_start_time": None,
"estimated_remaining_time": None,
"estimated_finish_time": None,
"heartbeat": None,
"accumulated_idle_time": 0.0,
}
def __init__(self, client):
self._client = client
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _load(self) -> dict:
val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
if val is None:
return dict(self._DEFAULTS)
return val
def _save(self, data: dict) -> None:
self._client.set_global_var(self._GLOBAL_VAR_KEY, data)
# ------------------------------------------------------------------
# Dict-like interface
# ------------------------------------------------------------------
def __getitem__(self, key):
return self._load()[key]
def __setitem__(self, key, value) -> None:
data = self._load()
data[key] = value
self._save(data)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self._load()!r})"
def get(self, key, default=None):
return self._load().get(key, default)
def update(self, *args, **kwargs) -> None:
"""Update multiple fields in a single round-trip."""
data = self._load()
data.update(*args, **kwargs)
self._save(data)
def reset(self) -> None:
"""Reset all progress fields to their default values."""
self._save(dict(self._DEFAULTS))
def as_dict(self) -> dict:
"""Return a plain copy of the current progress state."""
return self._load()
class LamNI(LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools):
def __init__(self, client):
super().__init__()
self.client = client
self.set_client(client)
self.device_manager = client.device_manager
self.align = XrayEyeAlign(client, self)
self.init = LaMNIInitStages(client)
# Correction state: corr_pos_x/y/angle (+ _2 iteration) are now
# global-var-backed properties defined in LamNIAlignmentMixin, so
# they are NOT initialized as plain lists here -- doing so would
# shadow the properties with instance attributes of the same name,
# silently breaking persistence. See lamni_alignment_mixin.py.
# Extracted collaborators
self.reconstructor = PtychoReconstructor(self.ptycho_reconstruct_foldername)
self.tomo_id_manager = TomoIDManager()
@@ -50,16 +131,17 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
self.special_angle_tolerance = 20
self._current_special_angles = []
# Progress tracking
self.progress = {}
self.progress["tomo_type"] = "Equally spaced sub-tomograms"
self.progress["subtomo"] = 0
self.progress["subtomo_projection"] = 0
self.progress["subtomo_total_projections"] = 1
self.progress["projection"] = 0
self.progress["total_projections"] = 1
self.progress["angle"] = 0
# Progress tracking, persisted via the BEC global variable
# "tomo_progress" so it survives a client restart -- deliberately NOT
# reset here: this proxy is constructed every time LamNI() is
# instantiated (i.e. every new client session), and resetting
# unconditionally on every instantiation would wipe tomo_start_time
# (and everything else) on exactly the restarts where recovering
# scan state matters most. A genuinely new tomo_scan() call resets
# the relevant fields itself (see the "new scan" branch in
# tomo_scan()); use tomo_progress_reset() to explicitly clear stale
# progress without starting a new scan.
self._progress_proxy = _ProgressProxy(self.client)
# ------------------------------------------------------------------
# Special angles
@@ -82,6 +164,124 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
self.special_angles = []
self.special_angle_repeats = 1
# ------------------------------------------------------------------
# Progress tracking
# ------------------------------------------------------------------
@property
def progress(self) -> _ProgressProxy:
"""Proxy dict backed by the BEC global variable ``tomo_progress``.
Readable from any BEC client session via::
client.get_global_var("tomo_progress")
Individual fields can be read and written just like a regular dict::
lamni.progress["projection"] # read
lamni.progress["projection"] = 42 # write (persists immediately)
To update multiple fields atomically use :py:meth:`_ProgressProxy.update`::
lamni.progress.update(projection=42, angle=90.0)
To reset all fields to their defaults::
lamni.progress.reset()
"""
return self._progress_proxy
@progress.setter
def progress(self, val: dict) -> None:
"""Replace the entire progress dict.
Accepts a plain :class:`dict` and persists it to the global var store.
"""
if not isinstance(val, dict):
raise TypeError(f"progress must be a dict, got {type(val).__name__!r}")
self._progress_proxy._save(val)
def tomo_progress_reset(self) -> None:
"""Explicitly clear the persisted tomo progress (start time, ETA,
current angle/subtomo/projection, accumulated idle time, ...).
Not called automatically on LamNI() startup -- that would wipe an
in-progress scan's state on every kernel restart. Call this by hand
if you want a clean progress display without it being tied to
starting a new tomo_scan() (which already resets the relevant
fields itself).
"""
self._progress_proxy.reset()
print("Tomo progress reset.")
@staticmethod
def _format_duration(seconds: float) -> str:
"""Format a duration in seconds as a human-readable string, e.g. '2h 03m 15s'."""
seconds = int(seconds)
h, remainder = divmod(seconds, 3600)
m, s = divmod(remainder, 60)
if h > 0:
return f"{h}h {m:02d}m {s:02d}s"
if m > 0:
return f"{m}m {s:02d}s"
return f"{s}s"
# ------------------------------------------------------------------
# X-ray eye alignment entry points
# ------------------------------------------------------------------
def xrayeye_alignment_start(self, keep_shutter_open: bool = False):
"""Run the BEC GUI-based X-ray eye alignment procedure.
Creates a fresh :class:`XrayEyeAlignGUI` instance, which resets the
correction state, and calls its ``align()`` method. The GUI window
is opened automatically. Interrupt with Ctrl-C to abort.
Args:
keep_shutter_open: If True the shutter is left open between angle
steps so the sample remains visible in live view.
"""
aligner = XrayEyeAlignGUI(self.client, self)
try:
aligner.align(keep_shutter_open=keep_shutter_open)
except KeyboardInterrupt as exc:
print("Alignment interrupted by user.")
try:
aligner.gui.hide_crosshair()
except Exception as gui_exc: # pylint: disable=broad-except
logger.warning(f"Failed to hide XRayEye alignment crosshair: {gui_exc}")
try:
aligner.gui.set_dap_params_forwarding(False)
except Exception as gui_exc: # pylint: disable=broad-except
logger.warning(f"Failed to disable XRayEye DAP parameter forwarding: {gui_exc}")
raise exc
# ── Reset manual shifts if needed ─────────────────────────────
mx = self.manual_shift_x
my = self.manual_shift_y
if mx != 0.0 or my != 0.0:
self.manual_shift_x = 0.0
self.manual_shift_y = 0.0
# Use OMNY-style green status message
self.OMNYTools.printgreen(
f"Manual shifts were reset to zero after X-ray eye alignment "
f"(previous values: x={mx:.3f}, y={my:.3f})."
)
def xrayeye_update_frame(self, keep_shutter_open: bool = False):
"""Capture a single fresh X-ray eye frame without running full alignment.
Useful for visually checking the sample position. Opens the X-ray eye
GUI if it is not already visible.
Args:
keep_shutter_open: If True the shutter is left open after the frame.
"""
aligner = XrayEyeAlignGUI(self.client, self)
aligner.update_frame(keep_shutter_open=keep_shutter_open)
# ------------------------------------------------------------------
# RT feedback / interferometer helpers
# ------------------------------------------------------------------
@@ -247,6 +447,23 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
def corridor_size(self, val: float):
self.client.set_global_var("corridor_size", val)
@property
def frames_per_trigger(self):
"""Number of burst frames acquired per point/projection. Used by
scans.lamni_fermat_scan (via tomo_scan_projection)."""
val = self.client.get_global_var("frames_per_trigger")
if val is None:
return 1
return val
@frames_per_trigger.setter
def frames_per_trigger(self, val: int):
if isinstance(val, bool) or not isinstance(val, int):
raise ValueError("frames_per_trigger must be a positive integer.")
if val <= 0:
raise ValueError("frames_per_trigger must be a positive integer.")
self.client.set_global_var("frames_per_trigger", val)
@property
def lamni_stitch_x(self):
val = self.client.get_global_var("lamni_stitch_x")
@@ -338,7 +555,9 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
@golden_projections_at_0_deg_for_damage_estimation.setter
def golden_projections_at_0_deg_for_damage_estimation(self, val: int):
self.client.set_global_var("golden_projections_at_0_deg_for_damage_estimation", val)
self.client.set_global_var(
"golden_projections_at_0_deg_for_damage_estimation", val
)
@property
def sample_name(self):
@@ -380,9 +599,7 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
)
def _write_tomo_scan_number(self, scan_number: int, angle: float, subtomo_number: int) -> None:
tomo_scan_numbers_file = os.path.expanduser(
"~/Data10/specES1/dat-files/tomography_scannumbers.txt"
)
tomo_scan_numbers_file = os.path.expanduser("~/data/raw/logs/tomography_scannumbers.txt")
with open(tomo_scan_numbers_file, "a+") as out_file:
out_file.write(
f"{scan_number} {angle} {dev.lsamrot.read()['lsamrot']['value']:.3f}"
@@ -390,7 +607,7 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
)
# ------------------------------------------------------------------
# Sample database — delegated to TomoIDManager in omny general tools
# Sample database — delegated to TomoIDManager
# ------------------------------------------------------------------
def add_sample_database(
@@ -414,9 +631,9 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
def tomo_scan_projection(self, angle: float):
scans = builtins.__dict__.get("scans")
additional_correction = self.align.compute_additional_correction(angle)
additional_correction_2 = self.align.compute_additional_correction_2(angle)
correction_xeye_mu = self.align.lamni_compute_additional_correction_xeye_mu(angle)
additional_correction = self.compute_additional_correction(angle)
additional_correction_2 = self.compute_additional_correction_2(angle)
correction_xeye_mu = self.lamni_compute_additional_correction_xeye_mu(angle)
self._current_scan_list = []
@@ -434,8 +651,8 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
stitch_x=stitch_x,
stitch_y=stitch_y,
stitch_overlap=self.tomo_stitch_overlap,
center_x=self.align.tomo_fovx_offset,
center_y=self.align.tomo_fovy_offset,
center_x=self.tomo_fovx_offset,
center_y=self.tomo_fovy_offset,
shift_x=(
self.manual_shift_x
+ correction_xeye_mu[0]
@@ -452,11 +669,12 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
angle=angle,
scan_type="fly",
exp_time=self.tomo_countingtime,
frames_per_trigger=self.frames_per_trigger,
optim_trajectory_corridor=corridor_size,
)
def tomo_reconstruct(self, base_path="~/Data10/specES1"):
"""Write the tomo reconstruct file for the reconstruction queue."""
def tomo_reconstruct(self, base_path="~/data/raw/logs/reconstruction_queue"):
"""write the tomo reconstruct file for the reconstruction queue"""
bec = builtins.__dict__.get("bec")
self.reconstructor.write(
scan_list=self._current_scan_list,
@@ -473,13 +691,39 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
# ------------------------------------------------------------------
def _print_progress(self):
# --- compute and store estimated remaining time -----------------------
start_str = self.progress.get("tomo_start_time")
projection = self.progress["projection"]
total = self.progress["total_projections"]
if start_str is not None and total > 0 and projection > 9:
now = datetime.datetime.now()
elapsed = (now - datetime.datetime.fromisoformat(start_str)).total_seconds()
# Exclude detected idle time (beamline-down pauses, a crash +
# restart gap, ...) so it doesn't make the scan look slower than
# it actually is while it's running.
elapsed -= self.progress.get("accumulated_idle_time", 0.0)
elapsed = max(elapsed, 1.0) # guard against a degenerate/negative denominator
rate = projection / elapsed # projections per second
remaining_s = (total - projection) / rate
self.progress["estimated_remaining_time"] = remaining_s
eta_str = self._format_duration(remaining_s)
finish_dt = now + datetime.timedelta(seconds=remaining_s)
self.progress["estimated_finish_time"] = finish_dt.isoformat()
finish_str = finish_dt.strftime("%Y-%m-%d %H:%M:%S")
else:
eta_str = "N/A"
finish_str = "N/A"
# ----------------------------------------------------------------------
print("\x1b[95mProgress report:")
print(f"Tomo type: ....................... {self.progress['tomo_type']}")
print(f"Projection: ...................... {self.progress['projection']}")
print(f"Total projections expected ....... {self.progress['total_projections']}")
print(f"Angle: ........................... {self.progress['angle']}")
print(f"Current subtomo: ................. {self.progress['subtomo']}")
print(f"Current projection within subtomo: {self.progress['subtomo_projection']}\x1b[0m")
print(f"Current projection within subtomo: {self.progress['subtomo_projection']}")
print(f"Estimated remaining time: ........ {eta_str}")
print(f"Estimated finish time: ........... {finish_str}\x1b[0m")
self._lamnigui_update_progress()
# ------------------------------------------------------------------
# Tomo scan orchestration
@@ -521,6 +765,34 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
successful = False
error_caught = False
if 0 <= angle < 360.05:
now = datetime.datetime.now()
prev_heartbeat_str = self.progress.get("heartbeat")
if prev_heartbeat_str is not None:
gap = (now - datetime.datetime.fromisoformat(prev_heartbeat_str)).total_seconds()
# Normal cadence between consecutive projections is roughly the
# acquisition time (times the number of stitch tiles per
# projection) plus motor/readout overhead. A gap well beyond
# that means something interrupted the scan in between
# (beamline-down interlock pause, a crash + manual restart,
# ...) -- attribute the excess to idle time so it doesn't drag
# down the apparent scan rate used for the ETA above. The
# 5x/60s margins are a heuristic, not a precise timing model --
# tune if it over/under-triggers in practice.
n_tiles = (2 * self.lamni_stitch_x + 1) * (2 * self.lamni_stitch_y + 1)
normal_cadence = max(
60.0, 5 * self.tomo_countingtime * self.frames_per_trigger * n_tiles
)
if gap > normal_cadence:
idle = gap - normal_cadence
self.progress["accumulated_idle_time"] = (
self.progress.get("accumulated_idle_time", 0.0) + idle
)
print(
f"Detected a {self._format_duration(gap)} gap since the last projection"
f" -- excluding {self._format_duration(idle)} from the ETA estimate."
)
self.progress["heartbeat"] = now.isoformat()
print(f"Starting LamNI scan for angle {angle} in subtomo {subtomo_number}")
self._print_progress()
while not successful:
@@ -550,10 +822,8 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
for scan_nr in range(start_scan_number, end_scan_number):
self._write_tomo_scan_number(scan_nr, angle, subtomo_number)
#todo here bl chk, if ok then successfull true
successful = True
def _golden(self, ii, howmany_sorted, maxangle=360, reverse=False):
"""Return the ii-th golden ratio angle within sorted bunches and its subtomo number."""
golden = []
@@ -628,6 +898,14 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
"BEC",
)
self.write_pdf_report()
self.progress["tomo_start_time"] = datetime.datetime.now().isoformat()
# reset stale estimates from any previous scan, otherwise the GUI
# would keep showing a leftover ETA from before this scan has
# accumulated enough projections to compute a fresh one
self.progress["estimated_remaining_time"] = None
self.progress["estimated_finish_time"] = None
self.progress["accumulated_idle_time"] = 0.0
self.progress["heartbeat"] = None
with scans.dataset_id_on_hold:
if self.tomo_type == 1:
@@ -716,6 +994,14 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
else:
raise ValueError(f"Unknown tomo_type: {self.tomo_type}.")
self.progress["projection"] = self.progress["total_projections"]
self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
self._print_progress()
print(
"Total measurement time lost to detected gaps:"
f" {self._format_duration(self.progress.get('accumulated_idle_time', 0.0))}"
)
# ------------------------------------------------------------------
# Parameter display and interactive update
# ------------------------------------------------------------------
@@ -733,11 +1019,13 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
print(f"Stitching overlap = {self.tomo_stitch_overlap}")
print(f"Circular FOV diam <microns> = {self.tomo_circfov}")
print(f"Reconstruction queue name = {self.ptycho_reconstruct_foldername}")
print("FOV offset rotates to find the ROI; manual shift moves the rotation center.")
print(f" _tomo_fovx_offset <mm> = {self.align.tomo_fovx_offset}")
print(f" _tomo_fovy_offset <mm> = {self.align.tomo_fovy_offset}")
print(f" _manual_shift_x <mm> = {self.manual_shift_x}")
print(f" _manual_shift_y <mm> = {self.manual_shift_y}")
print(f"Frames per trigger (burst) = {self.frames_per_trigger}")
print("FOV offset rotates to find the ROI; initial values determined in Xrayeye alignment.")
print("manual shift moves the rotation center.")
print(f" _tomo_fovx_offset <mm> = {self.tomo_fovx_offset:.4f}")
print(f" _tomo_fovy_offset <mm> = {self.tomo_fovy_offset:.4f}")
print(f" _manual_shift_x <mm> = {self.manual_shift_x:.4f}")
print(f" _manual_shift_y <mm> = {self.manual_shift_y:.4f}")
print("")
if self.tomo_type == 1:
print("\x1b[1mTomo type 1:\x1b[0m 8 equally spaced sub-tomograms (360 deg)")
@@ -764,8 +1052,8 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
print("Repeating projections at 0 deg at start of every second subtomogram.")
print(f"\nSample name: {self.sample_name}\n")
user_input = input("Are these parameters correctly set for your scan? ")
if user_input == "y":
if self.OMNYTools.yesno("Are these parameters correctly set for your scan?", "y"):
print("OK. continue.")
return
@@ -783,6 +1071,9 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
self.ptycho_reconstruct_foldername = self._get_val(
"Reconstruction queue", self.ptycho_reconstruct_foldername, str
)
self.frames_per_trigger = self._get_val(
"Frames per trigger (burst)", self.frames_per_trigger, int
)
print("Tomography type:")
print(" 1: 8 equally spaced sub-tomograms (360 deg)")
@@ -898,7 +1189,6 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
)
self.client.logbook.send_logbook_message(msg)
def get_calibration_of_capstops_left_and_right(self):
import time
print("""
@@ -909,28 +1199,28 @@ class LamNI(LamNIOpticsMixin, LamniGuiTools):
Example: At 0 deg, accessible rty -60 to 51. So the init was 5 microns off.
Then this routine here will provide data for the new capstop left and right.
""")
angle = 0
umv(dev.lsamrot,0)
umv(dev.lsamrot, 0)
print("Capstop right\nAngle, Voltage1, Voltage2")
mv(dev.lsamrot,361)
mv(dev.lsamrot, 361)
while angle <= 360:
angle = dev.lsamrot.readback.get()
voltage1=float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[1]"))
voltage2=float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[2]"))
if angle<360:
voltage1 = float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[1]"))
voltage2 = float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[2]"))
if angle < 360:
print(f"{angle},{voltage1},{voltage2}")
time.sleep(.3)
time.sleep(10)
print("\nCapstop left\nAngle, Voltage1, Voltage2")
mv(dev.lsamrot,-1)
mv(dev.lsamrot, -1)
while angle > 0:
angle = dev.lsamrot.readback.get()
voltage1=float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[1]"))
voltage2=float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[2]"))
if angle>0:
voltage1 = float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[1]"))
voltage2 = float(dev.lsamrot.controller.socket_put_and_receive("MG@AN[2]"))
if angle > 0:
print(f"{angle},{voltage1},{voltage2}")
time.sleep(.3)
print("Finished")
print("Finished")
csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_alignment_mixin.py
+369
View File
@@ -0,0 +1,369 @@
"""LamNI alignment correction infrastructure.
This mixin provides the correction infrastructure that is mixed directly into
the :class:`LamNI` class, mirroring the pattern used by
:class:`FlomniAlignmentMixin` in flomni.py.
The mixin assumes the hosting class provides:
- ``self.client`` (BECClient)
``corr_pos_x``/``corr_pos_y``/``corr_angle`` (+ the ``_2`` iteration) are
global-var-backed properties defined below -- the host class's ``__init__``
must NOT assign plain-list instance attributes of these names, as that would
shadow the properties and silently break persistence.
"""
from __future__ import annotations
import builtins
import os
import numpy as np
from bec_lib import bec_logger
from typeguard import typechecked
logger = bec_logger.logger
dev = builtins.__dict__.get("dev")
class LamNIAlignmentMixin:
"""Correction infrastructure for LamNI laminography.
Mixes into :class:`LamNI`. All methods are accessible directly as
``lamni.reset_correction()``, ``lamni.tomo_fovx_offset``, etc.,
matching the FlOMNI user API.
"""
# Pixel calibration: mm per pixel (0.2 mm FOV, 218 px)
PIXEL_CALIBRATION = 0.2 / 218
# ------------------------------------------------------------------
# Correction reset
# ------------------------------------------------------------------
def reset_correction(self):
"""Reset the look-up-table corrections to empty (iteration 1)."""
self.corr_pos_x = []
self.corr_pos_y = []
self.corr_angle = []
def reset_correction_2(self):
"""Reset the second iteration look-up-table correction to empty."""
self.corr_pos_x_2 = []
self.corr_pos_y_2 = []
self.corr_angle_2 = []
def reset_xray_eye_correction(self):
"""Delete the X-ray eye sinusoidal fit from the BEC global variable store."""
self.client.delete_global_var("tomo_fit_xray_eye")
# ------------------------------------------------------------------
# Additional lookup-table correction properties -- global-var-backed so
# they survive a BEC client restart, same pattern as tomo_fovx_offset/
# tomo_fovy_offset below and as FlomniAlignmentMixin's corr_pos_y/
# corr_angle_y on the FlOMNI side. Keys are "lamni_"-prefixed because the
# shape here differs from the FlOMNI side (real x *and* y correction
# plus a shared angle grid, vs. FlOMNI's y-only correction) -- not meant
# to be the same data, just the same persistence pattern.
#
# No default correction file exists for LamNI (unlike FlOMNI's
# default_correction_file): reset_correction()/reset_correction_2() just
# clear to empty, nothing is auto-loaded.
# ------------------------------------------------------------------
@property
def corr_pos_x(self):
val = self.client.get_global_var("lamni_corr_pos_x")
return [] if val is None else val
@corr_pos_x.setter
def corr_pos_x(self, val: list):
self.client.set_global_var("lamni_corr_pos_x", val)
@property
def corr_pos_y(self):
val = self.client.get_global_var("lamni_corr_pos_y")
return [] if val is None else val
@corr_pos_y.setter
def corr_pos_y(self, val: list):
self.client.set_global_var("lamni_corr_pos_y", val)
@property
def corr_angle(self):
val = self.client.get_global_var("lamni_corr_angle")
return [] if val is None else val
@corr_angle.setter
def corr_angle(self, val: list):
self.client.set_global_var("lamni_corr_angle", val)
@property
def corr_pos_x_2(self):
val = self.client.get_global_var("lamni_corr_pos_x_2")
return [] if val is None else val
@corr_pos_x_2.setter
def corr_pos_x_2(self, val: list):
self.client.set_global_var("lamni_corr_pos_x_2", val)
@property
def corr_pos_y_2(self):
val = self.client.get_global_var("lamni_corr_pos_y_2")
return [] if val is None else val
@corr_pos_y_2.setter
def corr_pos_y_2(self, val: list):
self.client.set_global_var("lamni_corr_pos_y_2", val)
@property
def corr_angle_2(self):
val = self.client.get_global_var("lamni_corr_angle_2")
return [] if val is None else val
@corr_angle_2.setter
def corr_angle_2(self, val: list):
self.client.set_global_var("lamni_corr_angle_2", val)
# ------------------------------------------------------------------
# FOV offset properties (backed by BEC global variable)
# ------------------------------------------------------------------
@property
def tomo_fovx_offset(self):
"""Horizontal FOV offset in mm (rotated with laminography geometry)."""
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):
"""Vertical FOV offset in mm (rotated with laminography geometry)."""
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])
# ------------------------------------------------------------------
# X-ray eye sinusoidal correction — read from files or GUI
# ------------------------------------------------------------------
# def read_xray_eye_correction(self, dir_path=None):
# """Load the sinusoidal X-ray eye fit from archived text files.
# Files are written by :meth:`XrayEyeAlign.write_output` at the end of
# every alignment run and are the fallback when the GUI has been closed.
# Args:
# dir_path: Directory containing ``ptychotomoalign_{A,B,C}{x,y}.txt``.
# Defaults to ``~/Data10/specES1/internal/``.
# """
# 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())
# print("New alignment parameters loaded from X-ray eye files:")
# self._print_xeye_fit(tomo_fit_xray_eye)
def read_xray_eye_correction_from_gui(self):
"""Load the sinusoidal X-ray eye fit from the live XRayEye GUI widget.
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
:meth:`lamni_compute_additional_correction_xeye_mu`.
.. important::
This method reads from the live GUI widget. If the XRayEye GUI
window has been closed since the alignment was performed the call
will fail. In that case use :meth:`read_xray_eye_correction` to
reload from the archived text files.
"""
_dev = builtins.__dict__.get("dev")
tomo_fit_xray_eye = np.zeros((2, 3))
params_x = _dev.omny_xray_gui.fit_params_x.get()
params_y = _dev.omny_xray_gui.fit_params_y.get()
for label, params in (("x", params_x), ("y", params_y)):
if (
not isinstance(params, dict)
or not {"SineModel_0_amplitude", "SineModel_0_shift", "LinearModel_1_intercept"}
<= params.keys()
):
raise RuntimeError(
f"No converged DAP sinusoidal fit available for the {label} axis "
f"(omny_xray_gui.fit_params_{label} did not return the expected "
"named parameters -- got "
f"{type(params).__name__ if not isinstance(params, dict) else 'a dict missing keys'}). "
"This means the fit hasn't converged / DAP forwarding hasn't produced "
"a result yet, not that the correction data is missing. Wait for the "
"fit to converge in the XRayEye GUI's Fit tab, or use "
"read_xray_eye_correction() to reload from the archived text files "
"instead."
)
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"]
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:")
self._print_xeye_fit(tomo_fit_xray_eye)
@staticmethod
def _print_xeye_fit(fit):
"""Pretty-print the 2×3 X-ray eye fit array."""
print(
f" X: A={fit[0][0]:.4f}, B={fit[0][1]:.4f}, C={fit[0][2]:.4f}\n"
f" Y: A={fit[1][0]:.4f}, B={fit[1][1]:.4f}, C={fit[1][2]:.4f}"
)
def lamni_compute_additional_correction_xeye_mu(self, angle):
"""Evaluate the sinusoidal X-ray eye correction at *angle* degrees.
Returns:
tuple: ``(correction_x_mm, correction_y_mm)``
"""
tomo_fit_xray_eye = self.client.get_global_var("tomo_fit_xray_eye")
if tomo_fit_xray_eye is None:
print("Not applying any X-ray eye correction. No fit data available.\n")
return (0, 0)
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:.6f} mm,"
f" y={correction_y:.6f} mm @ angle={angle}\n"
)
return (correction_x, correction_y)
# ------------------------------------------------------------------
# Additional look-up-table corrections (iteration 1 and 2)
# ------------------------------------------------------------------
def read_additional_correction(self, correction_file: str):
"""Load the iteration-1 correction lookup table from *correction_file*."""
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):
"""Load the iteration-2 correction lookup table from *correction_file*."""
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 containing ``corr_pos_x``, ``corr_pos_y``
and ``corr_angle`` entries.
Returns:
tuple: ``(corr_pos_x, corr_pos_y, corr_angle)`` as lists of floats.
"""
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 _ in range(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 the iteration-1 lookup-table correction for *angle*.
Returns:
tuple: ``(shift_x_mm, shift_y_mm)``
"""
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 the iteration-2 lookup-table correction for *angle*.
Returns:
tuple: ``(shift_x_mm, shift_y_mm)``
"""
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 {label}: x={shift_x}, y={shift_y}")
return (shift_x, shift_y)
csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py
+43 -3
View File
@@ -144,6 +144,8 @@ class LaMNIInitStages:
umv(dev.lsamrot, -1)
umv(dev.lsamrot, 0)
self.set_default_lamni_limits()
time.sleep(2)
dev.rtx.controller.feedback_disable_and_even_reset_lamni_angle_interferometer()
@@ -159,6 +161,38 @@ class LaMNIInitStages:
else:
return False
def set_default_lamni_limits(self):
"""
Apply safe, collision-protected default limits for LamNI.
Mirrors legacy SPEC limits.
"""
if not self.OMNYTools.yesno("Set default limits for LamNI?"):
print("Stopping.")
return
print("Setting LamNI limits...")
# Sample stages
dev.lsamx.limits = [6, 14]
dev.lsamy.limits = [6, 14]
dev.lsamrot.limits = [-3, 362]
# Optics (FZP)
dev.loptx.limits = [-1, -0.2]
dev.lopty.limits = [3.0, 3.6]
dev.loptz.limits = [82, 87]
# X-ray eye
dev.leyex.limits = [0, 25]
dev.leyey.limits = [0.5, 50]
# OSA / SmarAct
dev.losax.limits = [-1.5, 0.25]
dev.losay.limits = [-2.5, 4.1]
dev.losaz.limits = [-4.1, -0.5]
print("LamNI limits successfully applied.")
class LamNIOpticsMixin:
"""Optics movement methods: FZP, OSA, central stop and X-ray eye."""
@@ -178,12 +212,12 @@ class LamNIOpticsMixin:
epics_put("XOMNYI-XEYE-ACQ:0", 2)
umv(dev.lsamrot, 0)
umv(dev.dttrz, 5854, dev.fttrz, 2395)
#umv(dev.dttrz, 5854, dev.fttrz, 2395)
def leye_in(self):
bec.queue.next_dataset_number += 1
umv(dev.lsamrot, 0)
umv(dev.dttrz, 6419.677, dev.fttrz, 2959.979)
#umv(dev.dttrz, 6419.677, dev.fttrz, 2959.979)
while True:
moved_out = (input("Did the flight tube move out? (Y/n)") or "y").lower()
if moved_out == "y":
@@ -193,7 +227,7 @@ class LamNIOpticsMixin:
leyex_in = self._get_user_param_safe("leyex", "in")
leyey_in = self._get_user_param_safe("leyey", "in")
umv(dev.leyex, leyex_in, dev.leyey, leyey_in)
self.align.update_frame()
self.xrayeye_update_frame()
def _lfzp_in(self):
loptx_in = self._get_user_param_safe("loptx", "in")
@@ -208,7 +242,10 @@ class LamNIOpticsMixin:
self._lfzp_in()
if "rtx" in dev and dev.rtx.enabled:
print("Re-establishing interferometer feedback...")
_t0 = time.time()
dev.rtx.controller.feedback_enable_with_reset()
print(f"Interferometer feedback re-established ({time.time() - _t0:.1f} s).")
def loptics_in(self):
"""Move in the LamNI optics (FZP + OSA)."""
@@ -227,7 +264,10 @@ class LamNIOpticsMixin:
if "rtx" in dev and dev.rtx.enabled:
time.sleep(1)
print("Re-establishing interferometer feedback...")
_t0 = time.time()
dev.rtx.controller.feedback_enable_with_reset()
print(f"Interferometer feedback re-established ({time.time() - _t0:.1f} s).")
def lcs_in(self):
pass
csaxs_bec/bec_ipython_client/plugins/LamNI/test_correctionfile.txt
+25
View File
@@ -0,0 +1,25 @@
num_elements = 8
corr_pos_x[0] = 0.5
corr_pos_y[0] = -0.3
corr_angle[0] = 0.0
corr_pos_x[1] = 0.8
corr_pos_y[1] = 0.1
corr_angle[1] = 45.0
corr_pos_x[2] = 0.6
corr_pos_y[2] = 0.9
corr_angle[2] = 90.0
corr_pos_x[3] = -0.2
corr_pos_y[3] = 1.1
corr_angle[3] = 135.0
corr_pos_x[4] = -0.7
corr_pos_y[4] = 0.4
corr_angle[4] = 180.0
corr_pos_x[5] = -0.9
corr_pos_y[5] = -0.5
corr_angle[5] = 225.0
corr_pos_x[6] = -0.3
corr_pos_y[6] = -1.0
corr_angle[6] = 270.0
corr_pos_x[7] = 0.4
corr_pos_y[7] = -0.8
corr_angle[7] = 315.0
csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py
+569
View File
@@ -0,0 +1,569 @@
from __future__ import annotations
import builtins
import os
import time
from typing import TYPE_CHECKING
import h5py
import numpy as np
from bec_lib import bec_logger
logger = bec_logger.logger
bec = builtins.__dict__.get("bec")
dev = builtins.__dict__.get("dev")
scans = builtins.__dict__.get("scans")
def umv(*args):
return scans.umv(*args, relative=False)
if TYPE_CHECKING:
from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI
# Laminography alignment angles: full 360° in 45° steps
LAMNI_ALIGNMENT_ANGLES = [k * 45 for k in range(8)] # [0, 45, 90, 135, 180, 225, 270, 315]
class XrayEyeAlign:
"""BEC-GUI-based X-ray eye alignment for LamNI.
Replaces the old EPICS/LabView interface. Collects sample centre
coordinates (x and y) at 8 equally-spaced laminography angles over the
full 360°, submits them to the XRayEye widget's fit tab, and then reads
the resulting sinusoidal fit parameters back via
``lamni.read_xray_eye_correction_from_gui()``.
The key difference from the FlOMNI alignment is that LamNI needs *both*
x and y fits because the tilted rotation axis (61°) couples both
directions.
"""
# Pixel calibration in mm/pixel (multiply pixel values to get mm).
# The live value is read from the cam_xeye device's ``pixel_calibration``
# user parameter (see the ``pixel_calibration`` property, matching the
# FlOMNI side); this constant is only the fallback used when that
# device/parameter is unavailable.
PIXEL_CALIBRATION_DEFAULT = 0.2 / 218 # mm/pixel (0.2 mm FOV, 218 px)
PIXEL_CALIBRATION_USER_PARAM = "pixel_calibration"
def __init__(self, client, lamni: LamNI) -> None:
self.client = client
self.lamni = lamni
self.device_manager = client.device_manager
self.scans = client.scans
# Deliberately NOT calling self.lamni.reset_correction()/
# reset_xray_eye_correction() here: a fresh XrayEyeAlign is
# constructed on *every* call to xrayeye_alignment_start() AND
# xrayeye_update_frame() (see lamni.py) -- the latter is meant to be
# a lightweight "just grab a frame to look at" utility, not the start
# of a new alignment run. Now that corr_pos_x/y/angle and
# tomo_fit_xray_eye are persisted via BEC global vars, resetting them
# here would silently wipe a valid, already-fitted alignment every
# time someone calls xrayeye_update_frame() to check the sample.
# These resets belong in align() instead, matching what an actual
# fresh alignment run should do -- see the start of align() below.
# alignment_values[k] = [x_mm, y_mm]
# k=0 : FZP centre
# k=1 : sample at 0°, pre-shift (used only to compute shift_xy,
# the base shift to scan centre -- NOT one of the 8
# angle-fit points used in write_output())
# k=2 : sample at 0°, post-shift/at scan centre (first real
# angle-fit point; reference for all offsets below)
# k=3 : sample at 45°
# ...
# k=9 : sample at 315°
self.alignment_values: dict[int, list[float]] = {}
self.alignment_images = []
# ------------------------------------------------------------------
# GUI shortcut
# ------------------------------------------------------------------
@property
def gui(self):
"""Return the live XRayEye RPC handle stored by LamniGuiTools."""
return self.lamni.xeyegui
@property
def pixel_calibration(self) -> float:
"""Pixel calibration in mm/pixel.
Reads the ``pixel_calibration`` user parameter from the cam_xeye
device; falls back to ``PIXEL_CALIBRATION_DEFAULT`` if the device or
parameter is unavailable. Matches the FlOMNI side, which moved from a
hardcoded constant to this camera-sourced value.
"""
try:
mm_per_pixel = dev.cam_xeye.user_parameter.get(self.PIXEL_CALIBRATION_USER_PARAM)
except Exception:
mm_per_pixel = None
if mm_per_pixel is None:
return self.PIXEL_CALIBRATION_DEFAULT
return float(mm_per_pixel)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _reset_init_values(self):
self.shift_xy = [0.0, 0.0] # base shift to bring sample to beam centre [µm]
self._xray_fov_xy = [0.0, 0.0]
# Raw pixel coords + ROI size collected at each submit:
# [[step_k, x_px, y_px, w_px, h_px, image_idx], ...]
# image_idx refers to alignment_images[image_idx], i.e. the last
# frame captured before that submit (shutter is closed at submit time).
self.roi_pixel_data = []
def tomo_rotate(self, val: float):
umv(self.device_manager.devices.lsamrot, val)
def get_tomo_angle(self) -> float:
return self.device_manager.devices.lsamrot.readback.read()["lsamrot"]["value"]
def _disable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_disable()
def _enable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_enable_with_reset()
def update_frame(self, keep_shutter_open: bool = False):
"""Capture a fresh camera frame.
Args:
keep_shutter_open: If True the shutter is left open and the GUI
stays in live-view mode after the frame is taken. Useful when
it is hard to locate the sample between rotations. The caller
is responsible for closing the shutter afterwards.
"""
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
self.gui.on_live_view_enabled(True)
dev.fsh.fshopen()
time.sleep(1)
# store the image: the relevant frame for any submit that follows,
# since the shutter is closed again (unless keep_shutter_open) by
# the time the user actually clicks submit
self.alignment_images.append(dev.cam_xeye.get_last_image())
if not keep_shutter_open:
self.gui.on_live_view_enabled(False)
time.sleep(0.1)
dev.fsh.fshclose()
print("Received new frame.")
else:
print("Received new frame. Shutter remains open and live view active.")
def update_fov(self, k: int):
self._xray_fov_xy[0] = max(
getattr(dev.omny_xray_gui, f"width_x_{k}").get(), self._xray_fov_xy[0]
)
self._xray_fov_xy[1] = max(
getattr(dev.omny_xray_gui, f"width_y_{k}").get(), self._xray_fov_xy[1]
)
def movement_buttons_enabled(self, enable_x: bool, enable_y: bool):
self.gui.on_motors_enable(enable_x, enable_y)
def send_message(self, msg: str):
print(f"Alignment GUI: {msg}")
self.gui.user_message = msg
# ------------------------------------------------------------------
# Main alignment procedure
# ------------------------------------------------------------------
def _wait_for_dap_fit(self, timeout: float = 30.0, poll_interval: float = 1.0) -> bool:
"""Poll ``omny_xray_gui.fit_params_x``/``fit_params_y`` until both report
a converged DAP fit, or *timeout* seconds have passed.
Returns True if a converged fit was seen, False on timeout. Either way,
the caller (read_xray_eye_correction_from_gui) does its own check and
raises a clear error if the fit still isn't ready -- this just avoids
giving up after an arbitrary fixed sleep when the DAP service is simply
running a bit slower than usual.
"""
required_keys = {"SineModel_0_amplitude", "SineModel_0_shift", "LinearModel_1_intercept"}
t0 = time.time()
while time.time() - t0 < timeout:
params_x = dev.omny_xray_gui.fit_params_x.get()
params_y = dev.omny_xray_gui.fit_params_y.get()
if (
isinstance(params_x, dict)
and required_keys <= params_x.keys()
and isinstance(params_y, dict)
and required_keys <= params_y.keys()
):
print(f"DAP fit converged after {time.time() - t0:.1f} s.")
return True
time.sleep(poll_interval)
print(
f"DAP fit did not report converged parameters within {timeout:.0f} s -- "
"proceeding anyway; the next step will raise a clear error if it's "
"still not ready."
)
return False
def align(self, keep_shutter_open: bool = False):
"""Run the full LamNI X-ray eye alignment.
Args:
keep_shutter_open: If True the shutter is left open between angle
steps so the sample remains visible in live view. At the end
of the procedure the user is asked whether to close the shutter.
Matches the equivalent FlOMNI option.
Procedure
---------
Step 0 – FZP centre
Put FZP in, capture frame, user clicks FZP centre.
Step 1 – Sample at 0°
Put sample in, capture frame, user clicks sample centre.
Compute base shift (shift_xy) and move to scan centre.
Steps 2–8 – Sample at 45°, 90°, …, 315°
Rotate to each angle (via lamni_move_to_scan_center so the base
shift is applied), capture frame, user clicks sample centre.
After all submits
Compute x/y offsets, push to GUI fit tab, wait for DAP fit,
then load fit parameters into the global variable store.
"""
self.lamni.lamnigui_show_xeyealign()
self.gui.set_dap_params_forwarding(True)
self.send_message("Getting things ready. Please wait...")
# Actual start of a fresh alignment run -- see the comment in
# __init__ for why this can't live there.
self.lamni.reset_correction()
self.lamni.reset_xray_eye_correction()
self.gui.enable_submit_button(False)
# Initialise EPICS GUI device state
dev.omny_xray_gui.mvx.set(0)
dev.omny_xray_gui.mvy.set(0)
dev.omny_xray_gui.submit.set(0)
self.movement_buttons_enabled(False, False)
self._reset_init_values()
self.alignment_images = []
# --- Step 0: FZP centre ------------------------------------------
#self._disable_rt_feedback()
self.lamni.lfzp_in()
#self._enable_rt_feedback()
self.update_frame(keep_shutter_open)
self.gui.enable_submit_button(True)
dev.omny_xray_gui.step.set(0)
self.send_message("Submit centre of FZP.")
k = 0
while True:
if dev.omny_xray_gui.submit.get() == 1:
val_x = getattr(dev.omny_xray_gui, f"xval_x_{k}").get() * self.pixel_calibration
val_y = getattr(dev.omny_xray_gui, f"yval_y_{k}").get() * self.pixel_calibration
self.alignment_values[k] = [val_x, val_y]
print(
f"Clicked position {k}: "
f"x={self.alignment_values[k][0]:.4f} mm, "
f"y={self.alignment_values[k][1]:.4f} mm"
)
dev.omny_xray_gui.submit.set(0)
# Raw pixel position and ROI size at submit time. The
# relevant image is the last captured frame (shutter is
# closed by the time the user clicks submit).
_raw_x = getattr(dev.omny_xray_gui, f"xval_x_{k}").get()
_raw_y = getattr(dev.omny_xray_gui, f"yval_y_{k}").get()
_raw_w = getattr(dev.omny_xray_gui, f"width_x_{k}").get()
_raw_h = getattr(dev.omny_xray_gui, f"width_y_{k}").get()
_img_idx = len(self.alignment_images) - 1
print(
f" Submit k={k}: px x={_raw_x:.1f} y={_raw_y:.1f} "
f"w={_raw_w:.1f} h={_raw_h:.1f} img={_img_idx}"
)
self.roi_pixel_data.append([k, _raw_x, _raw_y, _raw_w, _raw_h, _img_idx])
# --- k=0: received FZP centre ----------------------------
if k == 0:
self.send_message("Please wait - moving sample in...")
self.movement_buttons_enabled(False, False)
self.gui.enable_submit_button(False)
self.lamni.loptics_out()
#self._disable_rt_feedback()
#time.sleep(0.3)
#self._enable_rt_feedback()
self.update_frame(keep_shutter_open)
# Mark the FZP centre on the live view: it stays visible
# as a fixed reference while the sample is aligned at
# each subsequent rotation angle below.
fzp_center_x = dev.omny_xray_gui.xval_x_0.get()
fzp_center_y = dev.omny_xray_gui.yval_y_0.get()
self.gui.set_crosshair_position(fzp_center_x, fzp_center_y)
self.gui.show_crosshair()
self.send_message("Find the sample and submit its centre.")
self.gui.enable_submit_button(True)
self.movement_buttons_enabled(True, True)
# --- k=1: sample at 0° - compute base shift --------------
elif k == 1:
# Displacement from FZP centre to sample centre gives the
# correction needed to bring the sample onto the beam axis.
# Sign conventions match the old alignment.py.
self.shift_xy[0] += (
self.alignment_values[0][0] - self.alignment_values[1][0]
) * 1000 # um
self.shift_xy[1] += (
self.alignment_values[1][1] - self.alignment_values[0][1]
) * 1000 # um
print(
f"Base shift: "
f"x={self.shift_xy[0]:.2f} um, "
f"y={self.shift_xy[1]:.2f} um"
)
self.send_message("Please wait - moving to scan centre...")
self.movement_buttons_enabled(False, False)
self.gui.enable_submit_button(False)
self.scans.lamni_move_to_scan_center(
shift_x=self.shift_xy[0] / 1000,
shift_y=self.shift_xy[1] / 1000,
angle=float(self.get_tomo_angle()),
).wait()
time.sleep(1)
self.scans.lamni_move_to_scan_center(
shift_x=self.shift_xy[0] / 1000,
shift_y=self.shift_xy[1] / 1000,
angle=float(self.get_tomo_angle()),
).wait()
dev.omny_xray_gui.angle.set(self.get_tomo_angle())
self.update_frame(keep_shutter_open)
self.send_message("Submit sample centre and FOV (0 deg).")
self.gui.enable_submit_button(True)
self.movement_buttons_enabled(True, False)
self.update_fov(k)
# --- k=2..8: sample at 45, 90, ..., 315 deg -------------
elif 1 < k <= 8:
self.send_message("Please wait - rotating...")
self.gui.enable_submit_button(False)
self.movement_buttons_enabled(False, False)
target_angle = LAMNI_ALIGNMENT_ANGLES[k - 1] # 45...315 deg
# Approach from halfway between previous and current angle
# to reduce mechanical hysteresis (same as old alignment.py)
prev_angle = LAMNI_ALIGNMENT_ANGLES[k - 2]
approach_angle = prev_angle + (target_angle - prev_angle) / 2
self._disable_rt_feedback()
self.tomo_rotate(approach_angle)
self.scans.lamni_move_to_scan_center(
shift_x=self.shift_xy[0] / 1000,
shift_y=self.shift_xy[1] / 1000,
angle=float(self.get_tomo_angle()),
).wait()
self._disable_rt_feedback()
self.tomo_rotate(target_angle)
self.scans.lamni_move_to_scan_center(
shift_x=self.shift_xy[0] / 1000,
shift_y=self.shift_xy[1] / 1000,
angle=float(self.get_tomo_angle()),
).wait()
dev.omny_xray_gui.angle.set(self.get_tomo_angle())
self.update_frame(keep_shutter_open)
self.send_message(f"Submit sample centre ({target_angle} deg).")
self.gui.enable_submit_button(True)
self.movement_buttons_enabled(True, False)
self.update_fov(k)
# --- k=9: all angles collected - finish ------------------
elif k == 9:
self.send_message("All angles collected - computing fit...")
self.gui.enable_submit_button(False)
self.movement_buttons_enabled(False, False)
self.update_fov(k)
self.gui.hide_crosshair()
break
k += 1
dev.omny_xray_gui.step.set(k)
# Handle live motor movement buttons.
# Only active during k=0 (FZP) and k=1 (sample at 0 deg) – both
# x and y movements accumulate into shift_xy and are applied via
# lamni_move_to_scan_center so the laminography coordinate
# transformation is correctly applied. This matches the old
# alignment.py exactly.
if k < 2:
_mvx = dev.omny_xray_gui.mvx.get()
_mvy = dev.omny_xray_gui.mvy.get()
if _mvx != 0 or _mvy != 0:
self.shift_xy[0] += _mvx
self.shift_xy[1] += _mvy
self.scans.lamni_move_to_scan_center(
shift_x=self.shift_xy[0] / 1000,
shift_y=self.shift_xy[1] / 1000,
angle=float(self.get_tomo_angle()),
).wait()
print(
f"Current centre: "
f"horizontal={self.shift_xy[0]:.2f} um, "
f"vertical={self.shift_xy[1]:.2f} um"
)
dev.omny_xray_gui.mvx.set(0)
dev.omny_xray_gui.mvy.set(0)
self.update_frame(keep_shutter_open)
time.sleep(0.2)
# ------------------------------------------------------------------
# All submits received - compute offsets and push to GUI fit tab
# ------------------------------------------------------------------
self.write_output()
fovx = self._xray_fov_xy[0] * self.pixel_calibration * 1000 / 2
fovy = self._xray_fov_xy[1] * self.pixel_calibration * 1000 / 2
print(
f"Largest FOV from X-ray eye alignment:\n"
f" fovx = {fovx:.0f} um, fovy = {fovy:.0f} um"
)
print(
f"Base shift (applied to all scan centres):\n"
f" x = {self.shift_xy[0]:.2f} um, y = {self.shift_xy[1]:.2f} um"
)
self.client.set_global_var("tomo_fov_offset", self.shift_xy)
# Switch GUI to fit tab and wait for DAP to finish fitting.
# A fixed sleep isn't reliable here -- how long the DAP service
# takes to fit and forward a result varies -- so poll the actual
# signal state instead of guessing a duration.
self.lamni.lamnigui_show_xeyealign_fittab()
print("Waiting for DAP sinusoidal fit to converge...")
self._wait_for_dap_fit(timeout=30)
# Read fit parameters from the GUI into the global variable store
print("Loading new alignment parameters from X-ray eye GUI fit.")
try:
self.lamni.read_xray_eye_correction_from_gui()
finally:
self.gui.set_dap_params_forwarding(False)
if keep_shutter_open:
answer = input("Close the shutter now? [Y/n]: ").strip().lower()
if answer in ("", "y", "yes"):
dev.fsh.fshclose()
self.gui.on_live_view_enabled(False)
print("Shutter closed.")
else:
print("Shutter left open.")
# Return to 0 deg
self._disable_rt_feedback()
self.tomo_rotate(0)
print(
"Done. You are ready to remove the X-ray eye and start ptychography scans.\n"
"Fine alignment: lamni.tomo_parameters() with offset 0, then lamni.sub_tomo_scan(1,0)"
)
# ------------------------------------------------------------------
# HDF5 output
# ------------------------------------------------------------------
def _save_alignment_data(self, file_path: str, fit_data: np.ndarray | None = None):
expanded = os.path.expanduser(file_path)
os.makedirs(os.path.dirname(expanded), exist_ok=True)
with h5py.File(expanded, "w") as f:
f.create_dataset(
"alignment_values", data=np.array(list(self.alignment_values.values()))
)
f.create_dataset("alignment_images", data=np.array(self.alignment_images))
if self.roi_pixel_data:
ds = f.create_dataset("roi_pixel_data", data=np.array(self.roi_pixel_data))
ds.attrs["columns"] = ["step_k", "x_px", "y_px", "w_px", "h_px", "image_idx"]
if fit_data is not None:
ds = f.create_dataset("alignment_fit", data=fit_data)
ds.attrs["rows"] = ["angles_deg", "fovx_offsets_um", "fovy_offsets_um"]
# ------------------------------------------------------------------
# Fit data preparation and submission
# ------------------------------------------------------------------
def write_output(self):
"""Compute x/y offsets for each angle and push them to the GUI fit tab.
Offsets are the displacement of the sample at each angle from the FZP
centre (k=0) -- matching FlOMNI's own reference point
(self.alignment_values[0] - self.alignment_values[k]). The FZP centre
is the only point in the whole procedure that's independent of
rotation angle, so it's the only valid zero-reference: using one of
the 8 angle points as the reference instead (as earlier versions of
this method did, first via an off-by-one at k=1, then via k=2) forces
that one point's own offset to be exactly zero by construction -- not
because the sample happens to be centred there, but because
subtracting a value from itself is always zero -- which biases the
whole sinusoidal fit rather than measuring it.
x_offset[i] = (x_at_FZP - x_at_angle_i) * 1000 [um]
y_offset[i] = (y_at_angle_i - y_at_FZP) * 1000 [um]
The array passed to submit_fit_array has shape (3, 8):
row 0: angles [0, 45, ..., 315]
row 1: x offsets [um]
row 2: y offsets [um]
Also writes a timestamped HDF5 file alongside the archival text file,
containing the full raw record of the alignment run: alignment_values
(FZP centre + all 8 angle clicks, in mm), alignment_images (one frame
per update_frame() call), roi_pixel_data (raw pixel coords/size at
each submit), and this same fit array as alignment_fit.
"""
# Archival text file (backward compatible with any external scripts)
file = os.path.expanduser("~/Data10/specES1/internal/xrayeye_alignmentvalues")
os.makedirs(os.path.dirname(file), exist_ok=True)
with open(file, "w") as f:
f.write("angle\thorizontal\tvertical\n")
for k in range(2, 10):
angle_deg = LAMNI_ALIGNMENT_ANGLES[k - 2]
x_off = (self.alignment_values[0][0] - self.alignment_values[k][0]) * 1000
y_off = (self.alignment_values[k][1] - self.alignment_values[0][1]) * 1000
f.write(f"{angle_deg}\t{x_off:.4f}\t{y_off:.4f}\n")
print(
f" Angle {angle_deg:3d} deg: "
f"x_offset={x_off:.2f} um, y_offset={y_off:.2f} um"
)
angles = np.array(LAMNI_ALIGNMENT_ANGLES, dtype=float)
x_offsets = np.array(
[
(self.alignment_values[0][0] - self.alignment_values[k][0]) * 1000
for k in range(2, 10)
]
)
y_offsets = np.array(
[
(self.alignment_values[k][1] - self.alignment_values[0][1]) * 1000
for k in range(2, 10)
]
)
data = np.array([angles, x_offsets, y_offsets])
# Timestamped HDF5 archive of the full raw alignment run
timestamp = time.strftime("%Y%m%d_%H%M%S")
file_h5 = f"~/data/raw/logs/xrayeye_alignmentvalues/xrayeye_alignmentvalues_{timestamp}.h5"
self._save_alignment_data(file_h5, fit_data=data)
# Push to XRayEye widget: feeds waveform_x (row 1) and waveform_y (row 2)
self.gui.submit_fit_array(data)
print(f"Fit data submitted with shape {data.shape}:\n{data}")
csaxs_bec/bec_ipython_client/plugins/flomni/docs/FZP 30nm Tomas Abe 2026.pdf
BIN
View File
Binary file not shown.
csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+6 -5
View File
@@ -46,10 +46,6 @@ def umv(*args):
return scans.umv(*args, relative=False)
class FlomniToolsError(Exception):
pass
class FlomniInitError(Exception):
pass
@@ -3765,7 +3761,12 @@ class Flomni(
csaxs_bec_basepath.parent / "bec_ipython_client" / "plugins" / "flomni" / logo_file_rel
).resolve()
print(logo_file)
bec.messaging.scilog.new().add_attachment(logo_file, width=200).add_text(
scilog = getattr(bec.messaging, "scilog", None)
if scilog is None or not getattr(scilog, "_enabled", False):
logger.warning("SciLog is not enabled; skipping PDF report entry.")
return
scilog.new().add_attachment(logo_file, width=200).add_text(
content.replace("\n", "<br>")
).add_tags("tomoscan").send()
csaxs_bec/bec_ipython_client/plugins/flomni/gui_tools.py
+1 -1
View File
@@ -43,7 +43,7 @@ class flomniGuiTools:
def set_client(self, client):
self.client = client
self.gui = self.client.gui
self.gui = getattr(client, "gui", None)
def flomnigui_show_gui(self):
if "flomni" in self.gui.windows:
csaxs_bec/bec_ipython_client/plugins/omny/omny_general_tools.py
+31 -32
View File
@@ -10,6 +10,7 @@ import termios
import threading
import time
import tty
import re
from pathlib import Path
import epics
@@ -323,11 +324,13 @@ class PtychoReconstructor:
with open(json_file, "w") as f:
json.dump(json_content, f, indent=4)
class TomoIDManager:
"""Registers a tomography measurement in the OMNY sample database
and returns its assigned tomo ID.
Falls back to tomo ID 0 for non-production accounts (e.g. test
accounts like "gac-x01dc") which the server rejects.
Usage:
id_manager = TomoIDManager()
tomo_id = id_manager.register(
@@ -341,11 +344,14 @@ class TomoIDManager:
)
"""
# OMNY_URL = "https://omny.web.psi.ch/samples/newmeasurement.php"
OMNY_URL = "https://v1p0zyg2w9n2k9c1.myfritz.net/samples/newmeasurement.php"
OMNY_USER = ""
OMNY_PASSWORD = ""
TMP_FILE = "~/currsamplesnr.txt"
FALLBACK_TOMO_ID = 0
@staticmethod
def _is_valid_eaccount(eaccount: str) -> bool:
"""True for real e-accounts (e.g. "e12345"), False for test accounts."""
return bool(re.fullmatch(r"e\d{5}", eaccount.strip()))
def register(
self,
@@ -359,18 +365,16 @@ class TomoIDManager:
) -> int:
"""Register a new measurement and return the assigned tomo ID.
Args:
sample_name (str): Name of the sample.
date (str): Date string (e.g. "2024-03-08").
eaccount (str): E-account identifier.
scan_number (int): First scan number of the measurement.
setup (str): Setup name (e.g. "lamni").
additional_info (str): Any additional sample information.
user (str): User name.
Returns:
int: The tomo ID assigned by the OMNY database.
Returns FALLBACK_TOMO_ID (0) if the account is not a real e-account
or if the server cannot be reached / returns an unusable response.
"""
if not self._is_valid_eaccount(eaccount):
logger.warning(
f"Account '{eaccount}' is not a valid e-account; "
f"skipping OMNY registration, using tomo ID {self.FALLBACK_TOMO_ID}."
)
return self.FALLBACK_TOMO_ID
url = (
f"{self.OMNY_URL}"
f"?sample={sample_name}"
@@ -381,25 +385,20 @@ class TomoIDManager:
f"&additional={additional_info}"
f"&user={user}"
)
# subprocess.run(
# f"wget --user={self.OMNY_USER} --password={self.OMNY_PASSWORD}"
# f" -q -O {self.TMP_FILE} '{url}'",
# shell=True,
# )
# print(url)
tmp_file = os.path.expanduser(self.TMP_FILE)
result = subprocess.run(f"wget -q -O {tmp_file} '{url}'", shell=True)
if result.returncode != 0:
raise OMNYToolsError(
f"wget failed (exit code {result.returncode}) fetching tomo ID from {self.OMNY_URL}"
)
try:
result = subprocess.run(f"wget -q -O {tmp_file} '{url}'", shell=True, timeout=30)
if result.returncode != 0:
raise OMNYToolsError(
f"wget failed (exit code {result.returncode}) fetching tomo ID from {self.OMNY_URL}"
)
with open(tmp_file) as f:
content = f.read().strip()
return int(content)
except FileNotFoundError as exc:
raise OMNYToolsError(f"wget did not produce output file {tmp_file}") from exc
except ValueError as exc:
raise OMNYToolsError(
f"Unexpected response from tomo ID server, got: {content!r}"
) from exc
except (subprocess.TimeoutExpired, FileNotFoundError, ValueError, OMNYToolsError) as exc:
logger.warning(
f"Could not obtain tomo ID from OMNY database ({exc}); "
f"falling back to tomo ID {self.FALLBACK_TOMO_ID}."
)
return self.FALLBACK_TOMO_ID
csaxs_bec/bec_ipython_client/plugins/omny/x_ray_eye_align.py
+363 -172
View File
@@ -5,233 +5,424 @@ import os
import time
from typing import TYPE_CHECKING
import numpy as np
from bec_lib import bec_logger
from csaxs_bec.bec_ipython_client.plugins.cSAXS import epics_get, epics_put, fshopen
logger = bec_logger.logger
# import builtins to avoid linter errors
bec = builtins.__dict__.get("bec")
dev = builtins.__dict__.get("dev")
scans = builtins.__dict__.get("scans")
def umv(*args):
return scans.umv(*args, relative=False)
if TYPE_CHECKING:
from bec_ipython_client.plugins.omny import OMNY
from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI
# Laminography alignment angles: full 360° in 45° steps
LAMNI_ALIGNMENT_ANGLES = [k * 45 for k in range(8)] # [0, 45, 90, 135, 180, 225, 270, 315]
class XrayEyeAlign:
# pixel calibration, multiply to get mm
PIXEL_CALIBRATION = 0.2 / 218 # .2 with binning
"""BEC-GUI-based X-ray eye alignment for LamNI.
def __init__(self, client, omny: OMNY) -> None:
Replaces the old EPICS/LabView interface. Collects sample centre
coordinates (x and y) at 8 equally-spaced laminography angles over the
full 360°, submits them to the XRayEye widget's fit tab, and then reads
the resulting sinusoidal fit parameters back via
``lamni.align.read_xray_eye_correction_from_gui()``.
The key difference from the FlOMNI alignment is that LamNI needs *both*
x and y fits because the tilted rotation axis (61°) couples both
directions.
"""
# Pixel calibration: multiply pixel coordinate by this to get mm
PIXEL_CALIBRATION = 0.2 / 218 # mm/pixel (0.2 mm FOV, 218 px)
def __init__(self, client, lamni: LamNI) -> None:
self.client = client
self.omny = omny
self.lamni = lamni
self.device_manager = client.device_manager
self.scans = client.scans
self.alignment_values = {}
self.omny.reset_correction()
self.omny.reset_tomo_alignment_fit()
# alignment_values[k] = [x_mm, y_mm]
# k=0 : FZP centre
# k=1 : sample at 0° (reference; shift_xy computed from k=0 vs k=1)
# k=2 : sample at 45°
# ...
# k=8 : sample at 315°
self.alignment_values: dict[int, list[float]] = {}
# Reset correction state via the existing alignment object on lamni
self.lamni.align.reset_correction()
self.lamni.align.reset_xray_eye_correction()
def _reset_init_values(self):
self.shift_xy = [0, 0]
self._xray_fov_xy = [0, 0]
def save_frame(self):
epics_put("XOMNYI-XEYE-SAVFRAME:0", 1)
def update_frame(self):
epics_put("XOMNYI-XEYE-ACQDONE:0", 0)
# start live
epics_put("XOMNYI-XEYE-ACQ:0", 1)
# wait for start live
while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0:
time.sleep(0.5)
print("waiting for live view to start...")
fshopen()
epics_put("XOMNYI-XEYE-ACQDONE:0", 0)
while epics_get("XOMNYI-XEYE-ACQDONE:0") == 0:
print("waiting for new frame...")
time.sleep(0.5)
time.sleep(0.5)
# stop live view
epics_put("XOMNYI-XEYE-ACQ:0", 0)
time.sleep(1)
# fshclose
print("got new frame")
def tomo_rotate(self, val: float):
# pylint: disable=undefined-variable
umv(self.device_manager.devices.osamroy, val)
def get_tomo_angle(self):
return self.device_manager.devices.osamroy.readback.get()
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])
# ------------------------------------------------------------------
# GUI shortcut
# ------------------------------------------------------------------
@property
def movement_buttons_enabled(self):
return [epics_get("XOMNYI-XEYE-ENAMVX:0"), epics_get("XOMNYI-XEYE-ENAMVY:0")]
def gui(self):
"""Return the live XRayEye RPC handle stored by LamniGuiTools."""
return self.lamni.xeyegui
@movement_buttons_enabled.setter
def movement_buttons_enabled(self, enabled: bool):
enabled = int(enabled)
epics_put("XOMNYI-XEYE-ENAMVX:0", enabled)
epics_put("XOMNYI-XEYE-ENAMVY:0", enabled)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _reset_init_values(self):
self.shift_xy = [0.0, 0.0] # base shift to bring sample to beam centre [µm]
self._xray_fov_xy = [0.0, 0.0]
def tomo_rotate(self, val: float):
umv(self.device_manager.devices.lsamrot, val)
def get_tomo_angle(self) -> float:
return self.device_manager.devices.lsamrot.readback.read()["lsamrot"]["value"]
def _disable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_disable()
def _enable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_enable_with_reset()
def update_frame(self, keep_shutter_open: bool = False):
"""Capture a fresh camera frame.
Args:
keep_shutter_open: If True the shutter is left open and the GUI
stays in live-view mode after the frame is taken. Useful when
it is hard to locate the sample between rotations. The caller
is responsible for closing the shutter afterwards.
"""
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
self.gui.on_live_view_enabled(True)
dev.omnyfsh.fshopen()
time.sleep(0.5)
if not keep_shutter_open:
self.gui.on_live_view_enabled(False)
time.sleep(0.1)
dev.omnyfsh.fshclose()
print("Received new frame.")
else:
print("Received new frame. Shutter remains open and live view active.")
def update_fov(self, k: int):
self._xray_fov_xy[0] = max(
getattr(dev.omny_xray_gui, f"width_x_{k}").get(), self._xray_fov_xy[0]
)
self._xray_fov_xy[1] = max(
getattr(dev.omny_xray_gui, f"width_y_{k}").get(), self._xray_fov_xy[1]
)
def movement_buttons_enabled(self, enable_x: bool, enable_y: bool):
self.gui.on_motors_enable(enable_x, enable_y)
def send_message(self, msg: str):
epics_put("XOMNYI-XEYE-MESSAGE:0.DESC", msg)
print(f"Alignment GUI: {msg}")
self.gui.user_message = msg
def align(self):
# reset shift xy and fov params
# ------------------------------------------------------------------
# Main alignment procedure
# ------------------------------------------------------------------
def align(self, keep_shutter_open: bool = False):
"""Run the full LamNI X-ray eye alignment.
Args:
keep_shutter_open: If True the shutter is left open between angle
steps so the sample remains visible in live view. At the end
of the procedure the user is asked whether to close the shutter.
Matches the equivalent FlOMNI option.
Procedure
---------
Step 0 – FZP centre
Put FZP in, capture frame, user clicks FZP centre.
Step 1 – Sample at 0°
Put sample in, capture frame, user clicks sample centre.
Compute base shift (shift_xy) and move to scan centre.
Steps 2–8 – Sample at 45°, 90°, …, 315°
Rotate to each angle (via lamni_move_to_scan_center so the base
shift is applied), capture frame, user clicks sample centre.
After all submits
Compute x/y offsets, push to GUI fit tab, wait for DAP fit,
then load fit parameters into the global variable store.
"""
self.lamni.lamnigui_show_xeyealign()
self.send_message("Getting things ready. Please wait...")
self.gui.enable_submit_button(False)
# Initialise EPICS GUI device state
dev.omny_xray_gui.mvx.set(0)
dev.omny_xray_gui.mvy.set(0)
dev.omny_xray_gui.submit.set(0)
self.movement_buttons_enabled(False, False)
self._reset_init_values()
self.tomo_rotate(0)
epics_put("XOMNYI-XEYE-ANGLE:0", 0)
# --- Step 0: FZP centre ------------------------------------------
self._disable_rt_feedback()
self.lamni.lfzp_in()
self._enable_rt_feedback()
self.omny.oeye_xray_in()
self.update_frame(keep_shutter_open)
self.omny.feedback_enable_with_reset()
# disable movement buttons
self.movement_buttons_enabled = False
sample_name = dev.omny_samples.get_sample_name_in_samplestage()
epics_put("XOMNYI-XEYE-SAMPLENAME:0.DESC", sample_name)
# this makes sure we are in a defined state
self.omny.feedback_disable()
epics_put("XOMNYI-XEYE-PIXELSIZE:0", self.PIXEL_CALIBRATION)
osamx_in = self.omny.OMNYTools._get_user_param_safe("osamx", "in")
umv(dev.osamx, osamx_in - 0.35)
self.omny.ofzp_in()
self.update_frame()
# enable submit buttons
self.movement_buttons_enabled = False
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
epics_put("XOMNYI-XEYE-STEP:0", 0)
self.send_message("Submit center value of FZP.")
self.gui.enable_submit_button(True)
dev.omny_xray_gui.step.set(0)
self.send_message("Submit centre of FZP.")
k = 0
while True:
if epics_get("XOMNYI-XEYE-SUBMIT:0") == 1:
val_x = epics_get(f"XOMNYI-XEYE-XVAL_X:{k}") / 2 * self.PIXEL_CALIBRATION # in mm
self.alignment_values[k] = val_x
print(f"Clicked position {k}: x {self.alignment_values[k]}")
rtx_position = dev.rtx.readback.get() / 1000
print(f"Current rtx position {rtx_position}")
self.alignment_values[k] -= rtx_position
print(f"Corrected position {k}: x {self.alignment_values[k]}")
if dev.omny_xray_gui.submit.get() == 1:
val_x = (
getattr(dev.omny_xray_gui, f"xval_x_{k}").get()
* self.PIXEL_CALIBRATION
)
val_y = (
getattr(dev.omny_xray_gui, f"yval_y_{k}").get()
* self.PIXEL_CALIBRATION
)
self.alignment_values[k] = [val_x, val_y]
print(
f"Clicked position {k}: "
f"x={self.alignment_values[k][0]:.4f} mm, "
f"y={self.alignment_values[k][1]:.4f} mm"
)
dev.omny_xray_gui.submit.set(0)
if k == 0: # received center value of FZP
self.send_message("please wait ...")
self.movement_buttons_enabled = False
epics_put("XOMNYI-XEYE-SUBMIT:0", -1) # disable submit button
# --- k=0: received FZP centre ----------------------------
if k == 0:
self.send_message("Please wait - moving sample in...")
self.movement_buttons_enabled(False, False)
self.gui.enable_submit_button(False)
self.omny.feedback_disable()
osamx_in = self.omny.OMNYTools._get_user_param_safe("osamx", "in")
umv(dev.osamx, osamx_in)
self.lamni.loptics_out()
self._disable_rt_feedback()
time.sleep(0.3)
self._enable_rt_feedback()
self.omny.ofzp_out()
self.update_frame(keep_shutter_open)
self.send_message("Find the sample and submit its centre.")
self.gui.enable_submit_button(True)
self.movement_buttons_enabled(True, True)
self.update_frame()
epics_put("XOMNYI-XEYE-RECBG:0", 1)
while epics_get("XOMNYI-XEYE-RECBG:0") == 1:
time.sleep(0.5)
print("waiting for background frame...")
# --- k=1: sample at 0° - compute base shift --------------
elif k == 1:
# Displacement from FZP centre to sample centre gives the
# correction needed to bring the sample onto the beam axis.
# Sign conventions match the old alignment.py.
self.shift_xy[0] += (
self.alignment_values[0][0] - self.alignment_values[1][0]
) * 1000 # um
self.shift_xy[1] += (
self.alignment_values[1][1] - self.alignment_values[0][1]
) * 1000 # um
print(
f"Base shift: "
f"x={self.shift_xy[0]:.2f} um, "
f"y={self.shift_xy[1]:.2f} um"
)
umv(dev.osamx, osamx_in)
time.sleep(0.5)
self.omny.feedback_enable_with_reset()
self.send_message("Please wait - moving to scan centre...")
self.movement_buttons_enabled(False, False)
self.gui.enable_submit_button(False)
self.update_frame()
self.send_message("Adjust sample height and submit center")
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
self.movement_buttons_enabled = True
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000,
self.shift_xy[1] / 1000,
self.get_tomo_angle(),
).wait()
time.sleep(1)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000,
self.shift_xy[1] / 1000,
self.get_tomo_angle(),
).wait()
elif 1 <= k < 5: # received sample center value at samroy 0 ... 315
self.send_message("please wait ...")
epics_put("XOMNYI-XEYE-SUBMIT:0", -1)
self.movement_buttons_enabled = False
umv(dev.rtx, 0)
self.tomo_rotate(k * 45)
epics_put("XOMNYI-XEYE-ANGLE:0", self.get_tomo_angle())
self.update_frame()
self.send_message("Submit sample center")
epics_put("XOMNYI-XEYE-SUBMIT:0", 0)
epics_put("XOMNYI-XEYE-ENAMVX:0", 1)
dev.omny_xray_gui.angle.set(self.get_tomo_angle())
self.update_frame(keep_shutter_open)
self.send_message("Submit sample centre and FOV (0 deg).")
self.gui.enable_submit_button(True)
self.movement_buttons_enabled(True, False)
self.update_fov(k)
elif k == 5: # received sample center value at samroy 270 and done
self.send_message("done...")
epics_put("XOMNYI-XEYE-SUBMIT:0", -1) # disable submit button
self.movement_buttons_enabled = False
# --- k=2..8: sample at 45, 90, ..., 315 deg -------------
elif 1 < k <= 8:
self.send_message("Please wait - rotating...")
self.gui.enable_submit_button(False)
self.movement_buttons_enabled(False, False)
target_angle = LAMNI_ALIGNMENT_ANGLES[k - 1] # 45...315 deg
# Approach from halfway between previous and current angle
# to reduce mechanical hysteresis (same as old alignment.py)
prev_angle = LAMNI_ALIGNMENT_ANGLES[k - 2]
approach_angle = prev_angle + (target_angle - prev_angle) / 2
self._disable_rt_feedback()
self.tomo_rotate(approach_angle)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000,
self.shift_xy[1] / 1000,
self.get_tomo_angle(),
).wait()
self._disable_rt_feedback()
self.tomo_rotate(target_angle)
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000,
self.shift_xy[1] / 1000,
self.get_tomo_angle(),
).wait()
dev.omny_xray_gui.angle.set(self.get_tomo_angle())
self.update_frame(keep_shutter_open)
self.send_message(f"Submit sample centre ({target_angle} deg).")
self.gui.enable_submit_button(True)
self.movement_buttons_enabled(True, False)
self.update_fov(k)
# --- k=9: all angles collected - finish ------------------
elif k == 9:
self.send_message("All angles collected - computing fit...")
self.gui.enable_submit_button(False)
self.movement_buttons_enabled(False, False)
self.update_fov(k)
break
k += 1
epics_put("XOMNYI-XEYE-STEP:0", k)
_xrayeyalignmvx = epics_get("XOMNYI-XEYE-MVX:0")
if _xrayeyalignmvx != 0:
umvr(dev.rtx, _xrayeyalignmvx)
print(f"Current rtx position {dev.rtx.readback.get() / 1000}")
epics_put("XOMNYI-XEYE-MVX:0", 0)
if k > 0:
epics_put(f"XOMNYI-XEYE-STAGEPOSX:{k}", dev.rtx.readback.get() / 1000)
time.sleep(3)
self.update_frame()
dev.omny_xray_gui.step.set(k)
# Handle live motor movement buttons.
# Only active during k=0 (FZP) and k=1 (sample at 0 deg) – both
# x and y movements accumulate into shift_xy and are applied via
# lamni_move_to_scan_center so the laminography coordinate
# transformation is correctly applied. This matches the old
# alignment.py exactly.
if k < 2:
# allow movements, store movements to calculate center
_xrayeyalignmvy = epics_get("XOMNYI-XEYE-MVY:0")
if _xrayeyalignmvy != 0:
self.omny.feedback_disable()
umvr(dev.osamy, _xrayeyalignmvy / 1000)
time.sleep(2)
epics_put("XOMNYI-XEYE-MVY:0", 0)
self.omny.feedback_enable_with_reset()
self.update_frame()
_mvx = dev.omny_xray_gui.mvx.get()
_mvy = dev.omny_xray_gui.mvy.get()
if _mvx != 0 or _mvy != 0:
self.shift_xy[0] += _mvx
self.shift_xy[1] += _mvy
self.scans.lamni_move_to_scan_center(
self.shift_xy[0] / 1000,
self.shift_xy[1] / 1000,
self.get_tomo_angle(),
).wait()
print(
f"Current centre: "
f"horizontal={self.shift_xy[0]:.2f} um, "
f"vertical={self.shift_xy[1]:.2f} um"
)
dev.omny_xray_gui.mvx.set(0)
dev.omny_xray_gui.mvy.set(0)
self.update_frame(keep_shutter_open)
time.sleep(0.2)
# ------------------------------------------------------------------
# All submits received - compute offsets and push to GUI fit tab
# ------------------------------------------------------------------
self.write_output()
fovx = self._xray_fov_xy[0] * self.PIXEL_CALIBRATION * 1000 / 2
fovy = self._xray_fov_xy[1] * self.PIXEL_CALIBRATION * 1000 / 2
self.tomo_rotate(0)
umv(dev.rtx, 0)
# free camera
epics_put("XOMNYI-XEYE-ACQ:0", 2)
print(
f"The largest field of view from the xrayeyealign was \nfovx = {fovx:.0f} microns, fovy"
f" = {fovy:.0f} microns"
f"Largest FOV from X-ray eye alignment:\n"
f" fovx = {fovx:.0f} um, fovy = {fovy:.0f} um"
)
print("Use the matlab routine to FIT the current alignment...")
print(
f"Base shift (applied to all scan centres):\n"
f" x = {self.shift_xy[0]:.2f} um, y = {self.shift_xy[1]:.2f} um"
)
self.client.set_global_var("tomo_fov_offset", self.shift_xy)
print("Then LOAD ALIGNMENT PARAMETERS by running omny.read_alignment_offset()\n")
# Switch GUI to fit tab and wait for DAP to finish fitting
self.lamni.lamnigui_show_xeyealign_fittab()
print("Waiting 5 s for DAP sinusoidal fit to converge...")
time.sleep(5)
# Read fit parameters from the GUI into the global variable store
print("Loading new alignment parameters from X-ray eye GUI fit.")
self.lamni.align.read_xray_eye_correction_from_gui()
if keep_shutter_open:
answer = input("Close the shutter now? [Y/n]: ").strip().lower()
if answer in ("", "y", "yes"):
dev.omnyfsh.fshclose()
self.gui.on_live_view_enabled(False)
print("Shutter closed.")
else:
print("Shutter left open.")
# Return to 0 deg and reset scan centre
self._disable_rt_feedback()
self.tomo_rotate(0)
umv(dev.rtx, 0)
print(
"Done. You are ready to remove the X-ray eye and start ptychography scans.\n"
"Fine alignment: lamni.tomo_parameters(), then lamni.tomo_alignment_scan()"
)
# ------------------------------------------------------------------
# Fit data preparation and submission
# ------------------------------------------------------------------
def write_output(self):
"""Compute x/y offsets for each angle and push them to the GUI fit tab.
Offsets are the displacement of the sample from its position at 0 deg
(k=1 is the reference):
x_offset[i] = (x_at_0deg - x_at_angle_i) * 1000 [um]
y_offset[i] = (y_at_angle_i - y_at_0deg) * 1000 [um]
The array passed to submit_fit_array has shape (3, 8):
row 0: angles [0, 45, ..., 315]
row 1: x offsets [um]
row 2: y offsets [um]
"""
# Archival text file (backward compatible with any external scripts)
file = os.path.expanduser("~/Data10/specES1/internal/xrayeye_alignmentvalues")
if not os.path.exists(file):
os.makedirs(os.path.dirname(file), exist_ok=True)
with open(file, "w") as alignment_values_file:
alignment_values_file.write("angle\thorizontal\n")
for k in range(1, 6):
fovx_offset = self.alignment_values[0] - self.alignment_values[k]
print(f"Writing to file new alignment: number {k}, value x {fovx_offset}")
alignment_values_file.write(f"{(k-1)*45}\t{fovx_offset*1000}\n")
os.makedirs(os.path.dirname(file), exist_ok=True)
with open(file, "w") as f:
f.write("angle\thorizontal\tvertical\n")
for k in range(1, 9):
angle_deg = LAMNI_ALIGNMENT_ANGLES[k - 1]
x_off = (
self.alignment_values[1][0] - self.alignment_values[k][0]
) * 1000
y_off = (
self.alignment_values[k][1] - self.alignment_values[1][1]
) * 1000
f.write(f"{angle_deg}\t{x_off:.4f}\t{y_off:.4f}\n")
print(
f" Angle {angle_deg:3d} deg: "
f"x_offset={x_off:.2f} um, y_offset={y_off:.2f} um"
)
angles = np.array(LAMNI_ALIGNMENT_ANGLES, dtype=float)
x_offsets = np.array(
[
(self.alignment_values[1][0] - self.alignment_values[k][0]) * 1000
for k in range(1, 9)
]
)
y_offsets = np.array(
[
(self.alignment_values[k][1] - self.alignment_values[1][1]) * 1000
for k in range(1, 9)
]
)
data = np.array([angles, x_offsets, y_offsets])
# Push to XRayEye widget: feeds waveform_x (row 1) and waveform_y (row 2)
self.gui.submit_fit_array(data)
print(f"Fit data submitted with shape {data.shape}:\n{data}")
csaxs_bec/device_configs/ptycho_lamni.yaml
+15 -1
View File
@@ -297,4 +297,18 @@ cam_xeye:
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: async
readoutPriority: async
userParameter:
pixel_calibration: 0.0009174311926605505 # mm/pixel (= 0.2 / 218)
############################################################
#################### GUI Signals ###########################
############################################################
omny_xray_gui:
description: Gui signals
deviceClass: csaxs_bec.devices.omny.xray_epics_gui.OMNYXRayAlignGUI
deviceConfig: {}
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
csaxs_bec/device_configs/simulated_omny/simulated_flomni.yaml
+12
View File
@@ -16,6 +16,18 @@
# omny_panda, omny_xray_gui.
############################################################
############################################################
#################### GUI Signals ###########################
############################################################
omny_xray_gui:
description: Gui signals
deviceClass: csaxs_bec.devices.omny.xray_epics_gui.OMNYXRayAlignGUI
deviceConfig: {}
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
############################################################
#################### flOMNI Galil motors ###################
############################################################
csaxs_bec/device_configs/simulated_omny/simulated_lamni.yaml
+15 -2
View File
@@ -136,7 +136,7 @@ lsamrot:
sign: 1
sim_stppermm: 50154.32099
sim_encpermm: 36000
sim_velocity: 12
sim_velocity: 48 #4x than hw
sim_initial_position: 0
deviceTags:
- lamni
@@ -328,7 +328,8 @@ cam_xeye:
onFailure: buffer
readOnly: false
readoutPriority: async
userParameter:
pixel_calibration: 0.0009174311926605505 # mm/pixel (= 0.2 / 218)
ddg1:
description: Simulated main delay generator for triggering
deviceClass: csaxs_bec.devices.sim.simulated_beamline_devices.SimulatedDDG1
@@ -347,3 +348,15 @@ fsh:
onFailure: raise
enabled: true
readoutPriority: monitored
############################################################
#################### GUI Signals ###########################
############################################################
omny_xray_gui:
description: Gui signals
deviceClass: csaxs_bec.devices.omny.xray_epics_gui.OMNYXRayAlignGUI
deviceConfig: {}
enabled: true
onFailure: buffer
readOnly: false
readoutPriority: on_request
csaxs_bec/devices/omny/xray_epics_gui.py
+2 -1
View File
@@ -13,7 +13,8 @@ class OMNYXRayAlignGUI(Device):
sample_name = Cpt(Signal, value=0)
angle = Cpt(Signal, value=0)
pixel_size = Cpt(Signal, value=0)
submit = Cpt(EpicsSignal, name="submit", read_pv="XOMNYI-XEYE-SUBMIT:0", auto_monitor=True)
submit = Cpt(Signal, value=0)
# submit = Cpt(EpicsSignal, name="submit", read_pv="XOMNYI-XEYE-SUBMIT:0", auto_monitor=True)
step = Cpt(Signal, value=0)
recbg = Cpt(Signal, value=0)
mvx = Cpt(Signal, value=0)
csaxs_bec/scans/LamNIFermatScan.py
+14 -10
View File
@@ -180,28 +180,32 @@ class LamNIMoveToScanCenter(RequestBase, LamNIMixin):
scan_name = "lamni_move_to_scan_center"
scan_report_hint = None
scan_type = "step"
required_kwargs = []
arg_input = {
"shift_x": ScanArgType.FLOAT,
"shift_y": ScanArgType.FLOAT,
"angle": ScanArgType.FLOAT,
}
arg_bundle_size = {"bundle": len(arg_input), "min": 1, "max": 1}
required_kwargs = ["shift_x", "shift_y", "angle"]
arg_input = {}
arg_bundle_size = {"bundle": 0, "min": 0, "max": 0}
def __init__(self, *args, parameter=None, **kwargs):
"""
Move LamNI to a new scan center.
Args:
*args: shift x, shift y, tomo angle in deg
shift_x (float): shift x in mm
shift_y (float): shift y in mm
angle (float): tomo angle in degrees
Examples:
>>> scans.lamni_move_to_scan_center(1.2, 2.8, 12.5)
>>> scans.lamni_move_to_scan_center(shift_x=1.2, shift_y=2.8, angle=12.5)
"""
super().__init__(parameter=parameter, **kwargs)
scan_kwargs = parameter.get("kwargs", {})
self.shift_x = float(scan_kwargs.get("shift_x", 0))
self.shift_y = float(scan_kwargs.get("shift_y", 0))
self.angle = float(scan_kwargs.get("angle", 0))
def run(self):
center_x, center_y = self._lamni_compute_scan_center(*self.caller_args)
center_x, center_y = self._lamni_compute_scan_center(
self.shift_x, self.shift_y, self.angle
)
yield from self.lamni_new_scan_center_interferometer(center_x, center_y)
docs/user/ptychography/lamni.md
+89 -65
View File
@@ -1,11 +1,12 @@
(user.ptychography.lamni)=
# LamNI
LamNI is an instrument for 3D ptychography via ptychographic X-ray computed lamninography (PyXL). The instrument is described in detail [here](https://www.dora.lib4ri.ch/psi/islandora/object/psi:33067).
LamNI is an instrument for 3D ptychography via ptychographic X-ray computed laminography (PyXL). The instrument is described in detail [here](https://www.dora.lib4ri.ch/psi/islandora/object/psi:33067).
## How to LamNI
## How to LamNI
… a step-by-step guide for _beamline staff and expert users_.
… a step-by-step guide for *beamline staff and expert users*.
### Sample change and alignment
@@ -17,70 +18,89 @@ Mount the new sample. The X-ray eye is already in, but the X-ray optics needs to
#### Coarse axis alignment
The effective position of the axis of rotation shifts with sample thickness or mounting position of the sample along the axis of rotation. The position of the axis of rotation is controlled by user parameters __center__ of the __lsamx__ and __lsamy__ stages. To observe the axis of rotation obtain the position of the Fresnel zone plate on the X-ray eye, possibly in the _ueye gui_ by:
The effective position of the axis of rotation shifts with sample thickness or mounting position of the sample along the axis of rotation. The position of the axis of rotation is controlled by user parameters **center** of the **lsamx** and **lsamy** stages. To observe the axis of rotation obtain the position of the Fresnel zone plate on the X-ray eye, possibly in the *ueye gui* by:
1. `lamni.lfzp_in()`, move the FZP in
1. `dev.rtx.controller.feedback_disable()`, disable feedback to allow lsam movements
1. `fshopen()`, open the shutter
1. `umv(dev.lsamrot,90)` to rotate the sample. One might observe the center of rotation at 0 and 180 degress.
1. `umvr(dev.lsamx,0.01)` to move lsamx and lsamy such that the center of rotation is at the center of the X-ray beam
1. `dev.lsamx` and `dev.lsamy` will print current position and the center value. Update the center value by
`dev.lsamx.update_user_parameter({'center':8.69})`
`dev.lsamy.update_user_parameter({'center':8.69})`
1. close the shutter: `dev.omnyfsh.fshclose()`
2. `dev.rtx.controller.feedback_disable()`, disable feedback to allow lsam movements
3. `fshopen()`, open the shutter
4. `umv(dev.lsamrot,90)` to rotate the sample. One might observe the center of rotation at 0 and 180 degrees.
5. `umvr(dev.lsamx,0.01)` to move lsamx and lsamy such that the center of rotation is at the center of the X-ray beam
6. `dev.lsamx` and `dev.lsamy` will print current position and the center value. Update the center value by
`dev.lsamx.update_user_parameter({'center':8.69})`
`dev.lsamy.update_user_parameter({'center':8.69})`
7. close the shutter: `dev.omnyfsh.fshclose()`
#### X-ray eye alignment
The GUI on the windows computer is used to obtain a coarse sample alignment. Start the alignment process (and clear any previous alignment) by
`lamni.align.align()`. With LamNI it can be very difficult to follow a region of interest as the sample rotates. Therefore the X-ray shutter will be open during the entire process. Therefore the windows software has to be set on __FORCE__ to continuously update frames and not freeze frames after rotation.
- run `SPEC_ptycho_align.m` (in matlab, use __force_ptychography = 0__)
- `lamni.align.read_xray_eye_correction()` to read the alignment parameters. The correction is based on sinusoidal fits in x and y direction. The values are computed by
`lamni_compute_additional_correction_xeye_mu(angle)`
- If slits were opened during alignment, close the slits `slits 1 to around 0.3`
- `lamni.leye_out()` remove the X-ray eye and move the flight tube in
- _possibly check slit0wh, idgap_
The BEC GUI is used to obtain a coarse sample alignment. Start the alignment process (which clears any previous alignment) by
`lamni.xrayeye_alignment_start()`
To only see one frame on the Windows GUI run `lamni.align.update_frame()`
This opens the X-ray eye widget automatically. The procedure collects the sample centre position at 8 angles (0°–315° in 45° steps, full 360° rotation). At each angle the user clicks the sample centre in the image and presses **Submit**. After all 8 angles the data is sent to the **Fit** tab of the GUI where a sinusoidal fit runs automatically in both x and y directions. The fit parameters are loaded automatically at the end of the procedure.
With LamNI it can be difficult to relocate the sample between rotations. To keep the shutter open throughout, pass:
`lamni.xrayeye_alignment_start(keep_shutter_open=True)`
To manually reload the fit parameters after the procedure has completed:
`lamni.read_xray_eye_correction_from_gui()`
**Note:** this reads from the live GUI widget via the `omny_xray_gui` device. It only works as long as the XRayEye GUI window remains open. If the window has been closed, reload from the archived text files instead:
`lamni.read_xray_eye_correction()`
(these files are written to `~/Data10/specES1/internal/xrayeye_alignmentvalues` at the end of every alignment run)
The correction is applied at each projection angle via
`lamni.lamni_compute_additional_correction_xeye_mu(angle)`
which is called automatically inside `lamni.tomo_scan_projection()`.
To capture a single fresh frame without running the full alignment:
`lamni.xrayeye_update_frame()`
or with the shutter left open: `lamni.xrayeye_update_frame(keep_shutter_open=True)`
* If slits were opened during alignment, close the slits: `slits 1` to around 0.3
* `lamni.leye_out()` remove the X-ray eye and move the flight tube in
* *possibly check slit0wh, idgap*
#### Fine alignment
The sample fine alignment can be obtained using ptychography. For this a short laminogram has to be recorded.
- `lamni.tomo_parameters()` adjust the parameters for a coarse scan: A large step size and large FOV. Especially select __FOV offset = 0__ and __number of projections = 96__ (only one sub-laminogram will be recorded).
- `lamni.sub_tomo_scan(1,0)` record one sub-laminogram
- use the corresponding scan numbers in `SPEC_ptycho_align.m`
- Record a last projection for all scans to reconstruct `lamni.tomo_scan_projection(0)` and wait for the reconstructions to be complete
- Run `SPEC_ptycho_align.m` (in Matlab, __force ptycho=1__, and __correct scan numbers__)
- Click the sample position in the Matlab GUI and then load the generated file by, for example
`lamni.align.read_additional_correction('/sls/X12SA/data/e20632/Data10/cxs_software/ptycho/correction_lamni_um_S05389_lamni_fit.txt')`
- With this alignment a second iteration could be performed. To read the second correction file use `lamni.align.read_additional_correction_2()`
* `lamni.tomo_parameters()` adjust the parameters for a coarse scan: A large step size and large FOV. Especially select **FOV offset = 0** and **number of projections = 96** (only one sub-laminogram will be recorded).
* `lamni.sub_tomo_scan(1,0)` record one sub-laminogram
* use the corresponding scan numbers in `SPEC_ptycho_align.m`
* Record a last projection for all scans to reconstruct `lamni.tomo_scan_projection(0)` and wait for the reconstructions to be complete
* Run `SPEC_ptycho_align.m` (in Matlab, **force ptycho=1**, and **correct scan numbers**)
* Click the sample position in the Matlab GUI and then load the generated file by, for example
`lamni.read_additional_correction('/sls/X12SA/data/e20632/Data10/cxs_software/ptycho/correction_lamni_um_S05389_lamni_fit.txt')`
* With this alignment a second iteration could be performed. To read the second correction file use `lamni.read_additional_correction_2()`
#### Shifting the FOV
- `lamni.align.tomo_fovx/y_offset=value` [mm] will shift the field of view. Perform this adjustment from projections collected at __lsamrot 0 degrees__.
* `lamni.tomo_fovx/y_offset=value` [mm] will shift the field of view. Perform this adjustment from projections collected at **lsamrot 0 degrees**. This shift will rotate. In contrast the manual shift will be a constant shift, identical at all angles.
### Laminography scan
Start the laminography scan by
1. `lamni.tomo_parameters()` adjust the parameters to the desired settings.
1. for test scans run
`lamni.tomo_scan_projection(angle)`
`lamni.tomo_reconstruct()`
1. `lamni.tomo_scan()` to start the lamninography scan
2. for test scans run
`lamni.tomo_scan_projection(angle)`
`lamni.tomo_reconstruct()`
3. `lamni.tomo_scan()` to start the laminography scan
### Tips and Tricks
#### Reset corrections
- `lamni.align.reset_correction()`
- `lamni.align.reset_correction_2()`
- `lamni.align.reset_xray_eye_correction()`
* `lamni.reset_correction()`
* `lamni.reset_correction_2()`
* `lamni.reset_xray_eye_correction()`
#### Adjusting beam size with feedback running
If the beam size needs to be changed with feedback running, e.g. to switch from near-field to far-field ptychography, following steps can be taken:
If the beam size needs to be changed with feedback running, e.g. to switch from near-field to far-field ptychography, following steps can be taken:
1. `dev.loptz.enable_set=True` to enable loptz movements with feedback running
1. `umvr(dev.lopz,_value_)` move loptz to the desired position.
1. `lamni._manual_shift_x/y = _value_` correct the stage run out from motion along the optical axis (units of um). The exact value can be checked by comparing feature positions in projections before/after adjusting loptz.
1. Potentially correct the alignment of the OSA
2. `umvr(dev.lopz,_value_)` move loptz to the desired position.
3. `lamni._manual_shift_x/y = _value_` correct the stage run out from motion along the optical axis (units of um). The exact value can be checked by comparing feature positions in projections before/after adjusting loptz.
4. Potentially correct the alignment of the OSA
#### BEC tips
@@ -97,8 +117,8 @@ This part of the manual describes the software structure in more detail.
The nano-positioning is controlled by a feedback loop running on a real-time linux based computer. With all related hardware connected, this loop has to be started manually.
1. Login to the computer by `ssh control@mpc2680`. The password is written on the physical machine.
1. `cd OMNY/lamni/`
1. `./startLAMNI`
2. `cd OMNY/lamni/`
3. `./startLAMNI`
Once the loop has started, it is possible to start bec with the LamNI configuration file.
@@ -123,60 +143,64 @@ The stages of LamNI are referenced in respect to their endswitches or reference
Show the status of all galil controllers (all stepper motors and the UPR rotation stage)
`dev.lsamx.controller.galil_show_all()`
The same holds true for the Smaract stages which control the OSA position. Their status can be checked by
The same holds true for the Smaract stages which control the OSA position. Their status can be checked by
`dev.losax.controller.show_all()`
In case referencing of the LamNI stages is required, run
`lamni.init.lamni_init_stages()`
This script will first verify that the stages are not in an initialized state, and then reference all stages in a safe way. The user will be warned in case of a potentially risky situation. This mainly involves a collision risk upstream with the exposure box exit window. It might be worth to check clearance prior to calling the init skript.
This script will first verify that the stages are not in an initialized state, and then reference all stages in a safe way. The user will be warned in case of a potentially risky situation. This mainly involves a collision risk upstream with the exposure box exit window. It might be worth to check clearance prior to calling the init script.
### Interferometer
The position feedback in LamNI is controlled in closed loop to an interferometric position measurement. To show the signal of the interferometers:
`lamni.show_signal_strength_interferometer()`
Typical values with proper alignment are
_TODO_
*TODO*
#### Interferometer feedback commands
- `dev.rtx.feedback_enable_with_reset()`
- `dev.rtx.feedback_disable()`
- `dev.rtx.feedback_enable_without_reset()` *is only used internally by lamni methods
- if reset of angle interferometer is required
`dev.rtx.feedback_disable_and_even_reset_lamni_angle_interferometer()`
- `dev.rtx.feedback_enable_with_reset()`
* `dev.rtx.feedback_enable_with_reset()`
* `dev.rtx.feedback_disable()`
* `dev.rtx.feedback_enable_without_reset()` \*is only used internally by lamni methods
* if reset of angle interferometer is required
`dev.rtx.feedback_disable_and_even_reset_lamni_angle_interferometer()`
* `dev.rtx.feedback_enable_with_reset()`
_ToDo Feedback status might be helpful. Plus make accessible via lamni.methods…_
*ToDo Feedback status might be helpful. Plus make accessible via lamni.methods…*
### Scanning in 2D and sample alignment
The underlying scan function can be called as
`scans.lamni_fermat_scan()`
Use `scans.lamni_fermat_scan?`for detailed information. A prerequisite for scanning is a running feedback system.
Use `scans.lamni_fermat_scan?` for detailed information. A prerequisite for scanning is a running feedback system.
### GUI tools
During operation the BEC GUI will show the relevant cameras or progress information. To manually switch view TAB completion on 'lamni.lamnigui_' will show all options to control the GUI. Most useful
'lamni.lamnigui_show_progress()' will show the measurement progress GUI
'lamnigui_show_xeyealign()' will show the XrayEye alignment GUI
During operation the BEC GUI will show the relevant cameras or progress information. TAB completion on `lamni.lamnigui_` will show all options to control the GUI. Most useful:
* `lamni.lamnigui_show_progress()` — shows the measurement progress with ring progress bar
* `lamni.lamnigui_show_xeyealign()` — opens the X-ray eye alignment GUI on the Alignment tab
* `lamni.lamnigui_show_xeyealign_fittab()` — opens the X-ray eye GUI on the Fit tab to inspect the sinusoidal fit
* `lamni.lamnigui_idle()` — shows the LamNI idle splash screen
* `lamni.lamnigui_docs()` — opens the PDF documentation viewer
### X-ray optics alignment
The positions of the optics stages are stored as stage parameters and are thus linked to the configuration file.
Example: The OSAx “in” position can be reviewed by `dev.losax.user_parameter`
Example: The OSAx "in" position can be reviewed by `dev.losax.user_parameter`
Update the value by (example "losax", "in") by `dev.losax.update_user_parameter({"in":value})`
`lamni.lfzp_info()` shows info about the available FZPs at the current energy of the beamline. Optional parameter is the photon _energy_ in keV.
`lamni.lfzp_info()` shows info about the available FZPs at the current energy of the beamline. Optional parameter is the photon *energy* in keV.
Example: `lamni.lfzp_info(6.2)`
The laser feedback will be disabled and fine alignment lost if foptx/y are moved!
Following functions exist to move the optics in and out, the naming is self-explaining.
- `lamni.lfzp_in()`
- `lamni.loptics_in()`
- `lamni.loptics_out()`
- `lamni.losa_in()`
- `lamni.losa_out()`
- `lamni.lfzp_in()`
* `lamni.lfzp_in()`
* `lamni.loptics_in()`
* `lamni.loptics_out()`
* `lamni.losa_in()`
* `lamni.losa_out()`
* `lamni.lfzp_in()`
docs/user/ptychography/omny.md
+1 -1
View File
@@ -214,7 +214,7 @@ This part of the manual describes the software structure in more detail.
The nano-positioning is controlled by a feedback loop running on a real-time linux based computer. With all related hardware connected, this loop has to be started manually.
1. Login to the computer by `ssh control@mpc3217`. The password is "engine".
1. Login to the computer by `ssh control@mpc3217`. The password is written on the physical machine.
1. `cd OMNY/OMNY/`
1. `./startOMNY`
tests/tests_bec_ipython_client/test_x_ray_eye_align.py
+44 -59
View File
@@ -2,7 +2,8 @@ from unittest import mock
from bec_lib.device import DeviceBase
from csaxs_bec.bec_ipython_client.plugins.LamNI import LamNI, XrayEyeAlign
from csaxs_bec.bec_ipython_client.plugins.LamNI import LamNI
from csaxs_bec.bec_ipython_client.plugins.LamNI.x_ray_eye_align import XrayEyeAlign
# pylint: disable=unused-import
@@ -12,6 +13,8 @@ from csaxs_bec.bec_ipython_client.plugins.LamNI import LamNI, XrayEyeAlign
# pylint: disable=redefined-outer-name
# pylint: disable=protected-access
XRAY_EYE_ALIGN = "csaxs_bec.bec_ipython_client.plugins.LamNI.x_ray_eye_align"
class RTControllerMock:
def feedback_disable(self):
@@ -26,56 +29,54 @@ class RTMock(DeviceBase):
enabled = True
def test_save_frame(bec_client_mock):
client = bec_client_mock
def _make_align(client):
client.device_manager.devices.xeye = DeviceBase(
name="xeye",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
)
lamni = LamNI(client)
align = XrayEyeAlign(client, lamni)
with mock.patch(
"csaxs_bec.bec_ipython_client.plugins.LamNI.alignment.epics_put"
) as epics_put_mock:
align.save_frame()
epics_put_mock.assert_called_once_with("XOMNYI-XEYE-SAVFRAME:0", 1)
return XrayEyeAlign(client, lamni)
def test_update_frame(bec_client_mock):
epics_put = "csaxs_bec.bec_ipython_client.plugins.LamNI.alignment.epics_put"
epics_get = "csaxs_bec.bec_ipython_client.plugins.LamNI.alignment.epics_get"
fshopen = "csaxs_bec.bec_ipython_client.plugins.LamNI.alignment.fshopen"
client = bec_client_mock
client.device_manager.devices.xeye = DeviceBase(
name="xeye",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
)
lamni = LamNI(client)
align = XrayEyeAlign(client, lamni)
with mock.patch(epics_put) as epics_put_mock:
with mock.patch(epics_get) as epics_get_mock:
with mock.patch(fshopen) as fshopen_mock:
align.update_frame()
epics_put_mock.assert_has_calls(
[
mock.call("XOMNYI-XEYE-ACQDONE:0", 0),
mock.call("XOMNYI-XEYE-ACQ:0", 1),
mock.call("XOMNYI-XEYE-ACQDONE:0", 0),
mock.call("XOMNYI-XEYE-ACQ:0", 0),
]
)
fshopen_mock.assert_called_once()
epics_get_mock.assert_called_with("XOMNYI-XEYE-ACQDONE:0")
align = _make_align(client)
align.lamni.xeyegui = mock.MagicMock()
dev_mock = mock.MagicMock()
dev_mock.cam_xeye.live_mode_enabled.get.return_value = False
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
align.update_frame()
dev_mock.cam_xeye.live_mode_enabled.put.assert_called_once_with(True)
dev_mock.fsh.fshopen.assert_called_once()
dev_mock.fsh.fshclose.assert_called_once()
dev_mock.cam_xeye.get_last_image.assert_called_once()
assert len(align.alignment_images) == 1
align.gui.on_live_view_enabled.assert_has_calls([mock.call(True), mock.call(False)])
def test_update_frame_keeps_shutter_open(bec_client_mock):
client = bec_client_mock
align = _make_align(client)
align.lamni.xeyegui = mock.MagicMock()
dev_mock = mock.MagicMock()
dev_mock.cam_xeye.live_mode_enabled.get.return_value = True
with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
align.update_frame(keep_shutter_open=True)
dev_mock.cam_xeye.live_mode_enabled.put.assert_not_called()
dev_mock.fsh.fshopen.assert_called_once()
dev_mock.fsh.fshclose.assert_not_called()
align.gui.on_live_view_enabled.assert_called_once_with(True)
def test_disable_rt_feedback(bec_client_mock):
client = bec_client_mock
client.device_manager.devices.xeye = DeviceBase(
name="xeye",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
)
lamni = LamNI(client)
align = XrayEyeAlign(client, lamni)
align = _make_align(client)
client.device_manager.devices.rtx = RTMock(
name="rtx",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
@@ -89,12 +90,7 @@ def test_disable_rt_feedback(bec_client_mock):
def test_enable_rt_feedback(bec_client_mock):
client = bec_client_mock
client.device_manager.devices.xeye = DeviceBase(
name="xeye",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
)
lamni = LamNI(client)
align = XrayEyeAlign(client, lamni)
align = _make_align(client)
client.device_manager.devices.rtx = RTMock(
name="rtx",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
@@ -107,23 +103,12 @@ def test_enable_rt_feedback(bec_client_mock):
def test_tomo_rotate(bec_client_mock):
import builtins
client = bec_client_mock
client._ip = mock.MagicMock()
client._update_namespace_callback = mock.MagicMock()
client.callbacks = mock.MagicMock()
client.load_high_level_interface("bec_hli")
client.device_manager.devices.xeye = DeviceBase(
name="xeye",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
)
lamni = LamNI(client)
align = XrayEyeAlign(client, lamni)
align = _make_align(client)
client.device_manager.devices.lsamrot = RTMock(
name="lsamrot",
config={"enabled": True, "deviceClass": "test_class", "readoutPriority": "baseline"},
)
with mock.patch.object(builtins, "umv") as umv:
with mock.patch(f"{XRAY_EYE_ALIGN}.umv") as umv:
align.tomo_rotate(5)
umv.assert_called_once_with(client.device_manager.devices.lsamrot, 5)
umv.assert_called_once_with(client.device_manager.devices.lsamrot, 5)