feat(lamni): refit the X-ray eye sine fit without bad clicks

lamni.xrayeye_refit([angles]) redoes the sinusoidal X-ray eye fit from the
last alignment's points, leaving out the given angles (misclicks, sample
barely visible). Prints each point's deviation from the new fit, shows the
kept points on the GUI Fit tab and asks before replacing tomo_fit_xray_eye.
The raw offsets are kept in the lamni_xeye_fit_data global var (cleared with
the fit); archived alignments can be refitted via file=.

Also clear omny_xray_gui.fit_params_x/y before submitting new alignment
points, so a previous run's fit is not taken as the new one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ALDnH16dC6iyHjhe5Ey4w9

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_<timestamp>.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."""

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)

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)