diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_alignment_mixin.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_alignment_mixin.py index a60d914f..f829f225 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_alignment_mixin.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_alignment_mixin.py @@ -17,6 +17,7 @@ from __future__ import annotations import builtins import os +import h5py import numpy as np from bec_lib import bec_logger from typeguard import typechecked @@ -25,6 +26,27 @@ logger = bec_logger.logger dev = builtins.__dict__.get("dev") +# Raw per-angle offsets of the most recent X-ray eye alignment, as written by +# XrayEyeAlign.write_output(): [angles_deg, x_offsets_um, y_offsets_um]. +XEYE_FIT_DATA_VAR = "lamni_xeye_fit_data" + + +def fit_xray_eye_sine(angles_deg, values): + """Least-squares fit of ``A * sin(radians(angle) + B) + C`` to *values*. + + Same model as the XRayEye GUI's DAP fit (SineModel with the frequency + fixed to 1 cycle/360 deg, plus a constant) and the same convention as + ``tomo_fit_xray_eye``. Written as ``a*sin + b*cos + C`` the model is + linear, so the fit is exact and needs no start values. + + Returns: + tuple: ``(A, B, C)`` with ``A >= 0`` and ``B`` in (-pi, pi]. + """ + theta = np.radians(np.asarray(angles_deg, dtype=float)) + design = np.column_stack([np.sin(theta), np.cos(theta), np.ones_like(theta)]) + (a, b, c), *_ = np.linalg.lstsq(design, np.asarray(values, dtype=float), rcond=None) + return float(np.hypot(a, b)), float(np.arctan2(b, a)), float(c) + class LamNIAlignmentMixin: """Correction infrastructure for LamNI laminography. @@ -54,8 +76,10 @@ class LamNIAlignmentMixin: self.corr_angle_2 = [] def reset_xray_eye_correction(self): - """Delete the X-ray eye sinusoidal fit from the BEC global variable store.""" + """Delete the X-ray eye sinusoidal fit, and the alignment points it was + fitted to, from the BEC global variable store.""" self.client.delete_global_var("tomo_fit_xray_eye") + self.client.delete_global_var(XEYE_FIT_DATA_VAR) # ------------------------------------------------------------------ # Additional lookup-table correction properties -- global-var-backed so @@ -241,6 +265,125 @@ class LamNIAlignmentMixin: print("New alignment parameters loaded from XRayEye GUI fit:") self._print_xeye_fit(tomo_fit_xray_eye) + def xrayeye_refit(self, exclude_angles=None, file: str | None = None, confirm: bool = True): + """Redo the X-ray eye sinusoidal fit, optionally leaving out angles. + + Use this after ``xrayeye_alignment_start()`` when one click is clearly + off the curve (sample barely visible at that angle, misclick, ...). + The clicks of the last alignment are refitted without the excluded + angles, a table of all points with their deviation from the new fit + is printed, and after confirmation the result replaces + ``tomo_fit_xray_eye``. Nothing is moved and no new frame is taken. + + Call it without arguments first to see the deviations of all points + from the full fit and spot the outlier. Calling it again with an + empty exclusion list restores the fit over all points. + + Args: + exclude_angles: angle or list of angles in deg (of 0, 45, ..., + 315) whose point is left out of the fit in both x and y. + file: refit an archived alignment instead of the last one, e.g. + ~/data/raw/logs/xrayeye_alignmentvalues/xrayeye_alignmentvalues_.h5 + confirm: ask before replacing the stored fit. + + Example: + lamni.xrayeye_refit() # inspect all points + lamni.xrayeye_refit([135]) # drop the 135 deg click + lamni.xrayeye_refit([135, 270]) + """ + data = self._load_xeye_fit_data(file) + angles = np.asarray(data[0], dtype=float) + offsets = {"x": np.asarray(data[1], dtype=float), "y": np.asarray(data[2], dtype=float)} + + if exclude_angles is None: + exclude_angles = [] + elif np.isscalar(exclude_angles): + exclude_angles = [exclude_angles] + excluded = np.zeros(len(angles), dtype=bool) + for angle in exclude_angles: + match = np.isclose(angles, float(angle) % 360) + if not match.any(): + raise ValueError( + f"No alignment point at {angle} deg. " + f"Available angles: {', '.join(f'{a:g}' for a in angles)}." + ) + excluded |= match + kept = ~excluded + # 3 free parameters per axis; with only 3 points any set of clicks + # fits exactly, so a bad click could no longer be noticed. + if kept.sum() < 4: + raise ValueError( + f"Only {kept.sum()} of {len(angles)} points left -- at least 4 are " + "needed for a meaningful fit." + ) + + fit = np.array([fit_xray_eye_sine(angles[kept], offsets[ax][kept]) for ax in ("x", "y")]) + residuals = {} + for i, ax in enumerate(("x", "y")): + amp, shift, const = fit[i] + residuals[ax] = offsets[ax] - (amp * np.sin(np.radians(angles) + shift) + const) + + print("X-ray eye points (offsets and deviation from the new fit, in um):") + print(" angle x_off x_dev y_off y_dev") + for j, angle in enumerate(angles): + flag = " excluded" if excluded[j] else "" + print( + f" {angle:5.0f} {offsets['x'][j]:9.2f} {residuals['x'][j]:8.2f}" + f" {offsets['y'][j]:9.2f} {residuals['y'][j]:8.2f}{flag}" + ) + for ax in ("x", "y"): + rms = np.sqrt(np.mean(residuals[ax][kept] ** 2)) + print(f" rms deviation of the fitted points, {ax}: {rms:.2f} um") + + print("New fit:") + self._print_xeye_fit(fit) + old_fit = self.client.get_global_var("tomo_fit_xray_eye") + if old_fit is not None: + print("Currently stored fit:") + self._print_xeye_fit(old_fit) + + self._show_xeye_refit_in_gui(angles[kept], offsets["x"][kept], offsets["y"][kept]) + + if confirm and not self.OMNYTools.yesno("Replace the stored X-ray eye fit?", "y"): + print("Stored fit left unchanged.") + return + self.client.set_global_var("tomo_fit_xray_eye", fit.tolist()) + if excluded.any(): + print( + "New X-ray eye fit stored (excluded: " + + ", ".join(f"{a:g}" for a in angles[excluded]) + + " deg)." + ) + else: + print("New X-ray eye fit stored (all points).") + + def _load_xeye_fit_data(self, file: str | None = None): + """Return the ``[angles, x_offsets, y_offsets]`` of the last alignment, + or of the archived alignment in *file*.""" + if file is not None: + with h5py.File(os.path.expanduser(file), "r") as f: + return f["alignment_fit"][()].tolist() + data = self.client.get_global_var(XEYE_FIT_DATA_VAR) + if data is None: + raise RuntimeError( + "No X-ray eye alignment points stored -- run lamni.xrayeye_alignment_start(), " + "or pass the archived alignment as file=... (see " + "~/data/raw/logs/xrayeye_alignmentvalues/)." + ) + return data + + def _show_xeye_refit_in_gui(self, angles, x_offsets, y_offsets): + """Show the points used in the refit on the XRayEye Fit tab. + + Display only: DAP parameter forwarding stays off, so the GUI's own fit + of these points does not touch the stored correction. + """ + try: + self.lamnigui_show_xeyealign_fittab() + self.xeyegui.submit_fit_array(np.array([angles, x_offsets, y_offsets])) + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"Could not show the refit points in the XRayEye GUI: {exc}") + @staticmethod def _print_xeye_fit(fit): """Pretty-print the 2×3 X-ray eye fit array.""" diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py index 56ef03f1..daa62265 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py @@ -11,6 +11,8 @@ import numpy as np from bec_lib import bec_logger +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_alignment_mixin import XEYE_FIT_DATA_VAR + logger = bec_logger.logger bec = builtins.__dict__.get("bec") @@ -602,6 +604,11 @@ class XrayEyeAlign: # ------------------------------------------------------------------ # All submits received - compute offsets and push to GUI fit tab # ------------------------------------------------------------------ + # Clear the previous run's fit parameters first: they'd otherwise + # satisfy _wait_for_dap_fit() immediately and be read back as this + # run's fit before DAP has fitted the new points. + dev.omny_xray_gui.fit_params_x.set({}) + dev.omny_xray_gui.fit_params_y.set({}) self.write_output() fovx = self._xray_fov_xy[0] * self.pixel_calibration * 1000 / 2 @@ -731,6 +738,8 @@ class XrayEyeAlign: 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) + # Raw offsets for lamni.xrayeye_refit(), which can leave out bad clicks + self.client.set_global_var(XEYE_FIT_DATA_VAR, data.tolist()) # Push to XRayEye widget: feeds waveform_x (row 1) and waveform_y (row 2) self.gui.submit_fit_array(data) diff --git a/tests/tests_bec_ipython_client/test_lamni_xrayeye_refit.py b/tests/tests_bec_ipython_client/test_lamni_xrayeye_refit.py new file mode 100644 index 00000000..8cd08eab --- /dev/null +++ b/tests/tests_bec_ipython_client/test_lamni_xrayeye_refit.py @@ -0,0 +1,165 @@ +"""lamni.xrayeye_refit(): redo the X-ray eye sine fit without bad clicks.""" + +import h5py +import numpy as np +import pytest + +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_alignment_mixin import ( + XEYE_FIT_DATA_VAR, + fit_xray_eye_sine, +) + +ANGLES = np.arange(8) * 45.0 +TRUE_X = (12.0, 0.7, -3.0) +TRUE_Y = (5.0, -2.1, 8.0) + + +def _model(params, angles=ANGLES): + amp, shift, const = params + return amp * np.sin(np.radians(angles) + shift) + const + + +class _FakeClient: + def __init__(self): + self._vars = {} + + def get_global_var(self, key): + return self._vars.get(key) + + def set_global_var(self, key, value): + self._vars[key] = value + + def delete_global_var(self, key): + self._vars.pop(key, None) + + +class _FakeTools: + def __init__(self, answer=True): + self.answer = answer + self.asked = 0 + + def yesno(self, *_args, **_kwargs): + self.asked += 1 + return self.answer + + +class _FakeXeyeGui: + def __init__(self): + self.submitted = None + + def submit_fit_array(self, data): + self.submitted = data + + +def make_lamni(data=None, answer=True) -> LamNI: + obj = object.__new__(LamNI) + obj.client = _FakeClient() + obj.OMNYTools = _FakeTools(answer) + obj.xeyegui = _FakeXeyeGui() + obj.lamnigui_show_xeyealign_fittab = lambda: None + if data is not None: + obj.client.set_global_var(XEYE_FIT_DATA_VAR, data) + return obj + + +def _data_with_outlier(angle_idx=3, dx=40.0, dy=-25.0): + x = _model(TRUE_X) + y = _model(TRUE_Y) + x[angle_idx] += dx + y[angle_idx] += dy + return [ANGLES.tolist(), x.tolist(), y.tolist()] + + +def test_fit_recovers_parameters(): + assert np.allclose(fit_xray_eye_sine(ANGLES, _model(TRUE_X)), TRUE_X) + + +def test_fit_negative_amplitude_is_equivalent(): + # A DAP fit may report -A with shift+pi; the function must be the same. + amp, shift, const = fit_xray_eye_sine(ANGLES, _model((-12.0, 0.7, -3.0))) + assert amp > 0 + assert np.allclose(_model((amp, shift, const)), _model((-12.0, 0.7, -3.0))) + + +def test_exclude_outlier_recovers_true_fit(): + lamni = make_lamni(_data_with_outlier()) + lamni.xrayeye_refit([135]) + fit = lamni.client.get_global_var("tomo_fit_xray_eye") + assert np.allclose(fit[0], TRUE_X) + assert np.allclose(fit[1], TRUE_Y) + # Only the kept points go to the GUI. + assert lamni.xeyegui.submitted.shape == (3, 7) + assert 135 not in lamni.xeyegui.submitted[0] + + +def test_scalar_and_wrapped_angle_accepted(): + lamni = make_lamni(_data_with_outlier(angle_idx=0)) + lamni.xrayeye_refit(360) + assert np.allclose(lamni.client.get_global_var("tomo_fit_xray_eye")[0], TRUE_X) + + +def test_no_exclusion_fits_all_points(): + lamni = make_lamni(_data_with_outlier()) + lamni.xrayeye_refit() + fit = lamni.client.get_global_var("tomo_fit_xray_eye") + assert not np.allclose(fit[0], TRUE_X) + assert lamni.xeyegui.submitted.shape == (3, 8) + + +def test_declined_leaves_stored_fit(): + lamni = make_lamni(_data_with_outlier(), answer=False) + lamni.client.set_global_var("tomo_fit_xray_eye", [[1, 2, 3], [4, 5, 6]]) + lamni.xrayeye_refit([135]) + assert lamni.OMNYTools.asked == 1 + assert lamni.client.get_global_var("tomo_fit_xray_eye") == [[1, 2, 3], [4, 5, 6]] + + +def test_confirm_false_does_not_ask(): + lamni = make_lamni(_data_with_outlier()) + lamni.xrayeye_refit([135], confirm=False) + assert lamni.OMNYTools.asked == 0 + assert lamni.client.get_global_var("tomo_fit_xray_eye") is not None + + +def test_unknown_angle_rejected(): + lamni = make_lamni(_data_with_outlier()) + with pytest.raises(ValueError, match="No alignment point at 100"): + lamni.xrayeye_refit([100]) + + +def test_too_few_points_rejected(): + lamni = make_lamni(_data_with_outlier()) + with pytest.raises(ValueError, match="at least 4"): + lamni.xrayeye_refit([0, 45, 90, 135, 180]) + + +def test_no_data_gives_clear_error(): + with pytest.raises(RuntimeError, match="xrayeye_alignment_start"): + make_lamni().xrayeye_refit() + + +def test_reset_clears_points(): + lamni = make_lamni(_data_with_outlier()) + lamni.reset_xray_eye_correction() + assert lamni.client.get_global_var(XEYE_FIT_DATA_VAR) is None + + +def test_refit_from_archive_file(tmp_path): + path = tmp_path / "xrayeye_alignmentvalues_20261013_120000.h5" + with h5py.File(path, "w") as f: + f.create_dataset("alignment_fit", data=np.array(_data_with_outlier())) + lamni = make_lamni() + lamni.xrayeye_refit([135], file=str(path)) + assert np.allclose(lamni.client.get_global_var("tomo_fit_xray_eye")[1], TRUE_Y) + + +def test_gui_failure_does_not_block_refit(): + lamni = make_lamni(_data_with_outlier()) + + def _boom(): + raise RuntimeError("GUI closed") + + lamni.lamnigui_show_xeyealign_fittab = _boom + lamni.xrayeye_refit([135]) + assert np.allclose(lamni.client.get_global_var("tomo_fit_xray_eye")[0], TRUE_X)