From 7863b8fd371bcd4e1f8cc24720bbd1b284d2b63b Mon Sep 17 00:00:00 2001 From: x01dc Date: Tue, 21 Jul 2026 12:40:45 +0200 Subject: [PATCH] feat/skip lfzp_in/losa_in/losa_out moves when already in position Ports flomni's ffzp_in()/fosa_in() "skip the move (and, for the FZP, the expensive feedback-disable + reset-enable cycle) if already there" optimization to lamni's lfzp_in()/losa_in(). Also adds the same skip-if-already-there check to losa_out(), which has no flomni equivalent to mirror (fosa_out() always moves unconditionally there) -- added as a lamni-side enhancement per discussion. Checked every existing caller of lfzp_in()/loptics_in() before making this change, since a skip changes behavior for callers that used to get an unconditional reset. Found one real risk: x_ray_eye_align.py's "Step 0: FZP centre" (the start of a fresh alignment run) had its manual feedback disable/enable calls already commented out, meaning it relied entirely on lfzp_in()'s old unconditional reset to establish a correct interferometer zero. flomni's equivalent call site doesn't have this problem because flomni's alignment flow does its own explicit feedback_enable_with_reset() independently beforehand -- lamni's doesn't. Fixed by passing force_feedback_reset=True at that one call site so it keeps its current always-reset behavior; no other lamni caller needed this. Item 4 of csaxs_bec/bec_ipython_client/plugins/LamNI/AI_docs/ FLOMNI_LAMNI_FEATURE_GAPS_2026-07.md. Co-Authored-By: Claude Sonnet 5 --- .../plugins/LamNI/lamni_optics_mixin.py | 103 +++++++++++-- .../plugins/LamNI/x_ray_eye_align.py | 8 +- .../test_lamni_optics_skip_if_in_position.py | 143 ++++++++++++++++++ 3 files changed, 244 insertions(+), 10 deletions(-) create mode 100644 tests/tests_bec_ipython_client/test_lamni_optics_skip_if_in_position.py diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py index 2c4b036..19196a3 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py @@ -1,6 +1,7 @@ import builtins import time +import numpy as np from rich import box from rich.console import Console from rich.table import Table @@ -232,21 +233,72 @@ class LamNIOpticsMixin: def _lfzp_in(self): loptx_in = self._get_user_param_safe("loptx", "in") lopty_in = self._get_user_param_safe("lopty", "in") - umv(dev.loptx, loptx_in, dev.lopty, lopty_in) - def lfzp_in(self): - """Move in the LamNI zone plate, disabling/re-enabling RT feedback around the move.""" - if "rtx" in dev and dev.rtx.enabled: + current_loptx = dev.loptx.readback.get() + current_lopty = dev.lopty.readback.get() + + tol = 0.003 + + # if either axis is outside the tolerance -> move both + need_move_optics = not np.isclose(current_loptx, loptx_in, atol=tol) or not np.isclose( + current_lopty, lopty_in, atol=tol + ) + + if need_move_optics: + umv(dev.loptx, loptx_in, dev.lopty, lopty_in) + else: + print("FZP is already at the in position.") + + return need_move_optics + + def lfzp_in(self, force_feedback_reset=False): + """ + Move in the LamNI zone plate. + This will disable rt feedback, move the FZP and re-enable the feedback. + + The FZP move requires rt feedback OFF, and moving the FZP invalidates + the interferometer zero, so feedback is re-enabled *with reset* + afterwards. That reset is expensive: it re-zeros the interferometers + and moves you away from wherever the sample currently sits, which is + undesirable when the FZP is already in and feedback is already running + (e.g. repeated alignment scans, or interleaving an alignment run into a + tomogram). + + Therefore the disable/move/reset cycle is skipped entirely when the FZP + does not actually need to move. Pass ``force_feedback_reset=True`` to + force the full disable + reset cycle even if the FZP is already in -- + e.g. x_ray_eye_align.py's "Step 0: FZP centre" of a fresh alignment + run always needs a freshly-zeroed interferometer reference, + regardless of whether the FZP physically needs to move. + """ + rtx_present = "rtx" in dev and dev.rtx.enabled + + # Only disable feedback if we're going to move the FZP (or a reset was + # explicitly requested). If the FZP is already in and feedback is + # already running, leave it untouched -- disabling and + # re-enabling-with-reset would needlessly re-zero the interferometers. + needs_move = not self._lfzp_is_in() + do_cycle = needs_move or force_feedback_reset + + if rtx_present and do_cycle: dev.rtx.controller.feedback_disable() self._lfzp_in() - if "rtx" in dev and dev.rtx.enabled: + if rtx_present and do_cycle: 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 _lfzp_is_in(self, tol=0.003): + """True if both FZP axes (loptx, lopty) are within ``tol`` of their IN position.""" + loptx_in = self._get_user_param_safe("loptx", "in") + lopty_in = self._get_user_param_safe("lopty", "in") + return np.isclose(dev.loptx.readback.get(), loptx_in, atol=tol) and np.isclose( + dev.lopty.readback.get(), lopty_in, atol=tol + ) + def loptics_in(self): """Move in the LamNI optics (FZP + OSA).""" self.lfzp_in() @@ -279,14 +331,47 @@ class LamNIOpticsMixin: losax_in = self._get_user_param_safe("losax", "in") losay_in = self._get_user_param_safe("losay", "in") losaz_in = self._get_user_param_safe("losaz", "in") - umv(dev.losax, losax_in, dev.losay, losay_in) - umv(dev.losaz, losaz_in) + + current_losax = dev.losax.readback.get() + current_losay = dev.losay.readback.get() + current_losaz = dev.losaz.readback.get() + + tol = 0.003 + + need_move_osa = ( + not np.isclose(current_losax, losax_in, atol=tol) + or not np.isclose(current_losay, losay_in, atol=tol) + or not np.isclose(current_losaz, losaz_in, atol=tol) + ) + + if need_move_osa: + umv(dev.losax, losax_in, dev.losay, losay_in) + umv(dev.losaz, losaz_in) + else: + print("OSA is already at the IN position.") def losa_out(self): losay_out = self._get_user_param_safe("losay", "out") losaz_out = self._get_user_param_safe("losaz", "out") - umv(dev.losaz, losaz_out) - umv(dev.losay, losay_out) + + current_losay = dev.losay.readback.get() + current_losaz = dev.losaz.readback.get() + + tol = 0.003 + + # No flomni equivalent for OSA-out exists (fosa_out() always moves + # unconditionally there) -- added here anyway as a lamni-side + # enhancement, mirroring the same skip-if-already-there shape as + # losa_in()/lfzp_in() above. + need_move_osa = not np.isclose(current_losay, losay_out, atol=tol) or not np.isclose( + current_losaz, losaz_out, atol=tol + ) + + if need_move_osa: + umv(dev.losaz, losaz_out) + umv(dev.losay, losay_out) + else: + print("OSA is already at the OUT position.") def lfzp_info(self, mokev_val=-1): if mokev_val == -1: 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 2b12b35..ebaffc4 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 @@ -255,7 +255,13 @@ class XrayEyeAlign: # --- Step 0: FZP centre ------------------------------------------ #self._disable_rt_feedback() - self.lamni.lfzp_in() + # force_feedback_reset=True: this is the start of a fresh alignment + # run, so the interferometer reference must always be freshly + # re-zeroed here, regardless of whether the FZP happens to already + # be in position (e.g. left over from a previous run/tomogram) -- + # lfzp_in()'s default skip-if-already-in-position optimization must + # not apply at this specific call site. + self.lamni.lfzp_in(force_feedback_reset=True) #self._enable_rt_feedback() self.update_frame(keep_shutter_open) diff --git a/tests/tests_bec_ipython_client/test_lamni_optics_skip_if_in_position.py b/tests/tests_bec_ipython_client/test_lamni_optics_skip_if_in_position.py new file mode 100644 index 0000000..ac9f9e7 --- /dev/null +++ b/tests/tests_bec_ipython_client/test_lamni_optics_skip_if_in_position.py @@ -0,0 +1,143 @@ +"""Tests for lamni's "skip if already in position" optics optimization, +ported from flomni's ffzp_in()/fosa_in() (see csaxs_bec/bec_ipython_client/ +plugins/LamNI/AI_docs/FLOMNI_LAMNI_FEATURE_GAPS_2026-07.md, item 4). + +Methods under test (lamni_optics_mixin.py) reference module-level `dev`/ +`umv` globals (resolved from this module's own namespace, not an instance +attribute), so tests monkeypatch those module globals directly. +""" + +import types + +import pytest + +import csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_optics_mixin as optics_mixin_module +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_optics_mixin import LamNIOpticsMixin + + +class FakeAxis: + def __init__(self, value, in_pos=None, out_pos=None): + self.value = value + self.readback = types.SimpleNamespace(get=lambda: self.value) + self.user_parameter = {} + if in_pos is not None: + self.user_parameter["in"] = in_pos + if out_pos is not None: + self.user_parameter["out"] = out_pos + + +class FakeController: + def __init__(self): + self.calls = [] + + def feedback_disable(self): + self.calls.append("disable") + + def feedback_enable_with_reset(self): + self.calls.append("enable_with_reset") + + +class FakeDev: + def __init__(self, rtx_enabled=True): + self.loptx = FakeAxis(-0.5, in_pos=-0.5, out_pos=-1.0) + self.lopty = FakeAxis(3.3, in_pos=3.3, out_pos=3.0) + self.losax = FakeAxis(-1.0, in_pos=-1.0, out_pos=0.0) + self.losay = FakeAxis(-0.2, in_pos=-0.2, out_pos=1.0) + self.losaz = FakeAxis(1.0, in_pos=1.0, out_pos=-1.0) + self.rtx = types.SimpleNamespace(enabled=rtx_enabled, controller=FakeController()) + + def __contains__(self, name): + return hasattr(self, name) + + def __getitem__(self, name): + return getattr(self, name) + + +@pytest.fixture +def fake_dev(monkeypatch): + dev = FakeDev() + monkeypatch.setattr(optics_mixin_module, "dev", dev) + return dev + + +@pytest.fixture +def umv_calls(monkeypatch): + calls = [] + monkeypatch.setattr(optics_mixin_module, "umv", lambda *args: calls.append(args)) + return calls + + +def make_optics(): + return LamNIOpticsMixin() + + +def test_lfzp_in_skips_move_when_already_in_position(fake_dev, umv_calls, capsys): + optics = make_optics() + + optics.lfzp_in() + + assert umv_calls == [] + assert "already at the in position" in capsys.readouterr().out + assert fake_dev.rtx.controller.calls == [] + + +def test_lfzp_in_moves_and_resets_feedback_when_out_of_position(fake_dev, umv_calls): + fake_dev.loptx.value = -0.9 # not at "in" (-0.5) + optics = make_optics() + + optics.lfzp_in() + + assert umv_calls == [(fake_dev.loptx, -0.5, fake_dev.lopty, 3.3)] + assert fake_dev.rtx.controller.calls == ["disable", "enable_with_reset"] + + +def test_lfzp_in_force_feedback_reset_runs_cycle_even_if_already_in(fake_dev, umv_calls): + optics = make_optics() + + optics.lfzp_in(force_feedback_reset=True) + + assert umv_calls == [] # still no physical move needed + assert fake_dev.rtx.controller.calls == ["disable", "enable_with_reset"] + + +def test_losa_in_skips_move_when_already_in_position(fake_dev, umv_calls, capsys): + optics = make_optics() + + optics.losa_in() + + assert umv_calls == [] + assert "already at the IN position" in capsys.readouterr().out + + +def test_losa_in_moves_when_out_of_position(fake_dev, umv_calls): + fake_dev.losaz.value = 5.0 # not at "in" (1.0) + optics = make_optics() + + optics.losa_in() + + assert umv_calls == [ + (fake_dev.losax, -1.0, fake_dev.losay, -0.2), + (fake_dev.losaz, 1.0), + ] + + +def test_losa_out_skips_move_when_already_out(fake_dev, umv_calls, capsys): + fake_dev.losay.value = 1.0 # matches "out" + fake_dev.losaz.value = -1.0 # matches "out" + optics = make_optics() + + optics.losa_out() + + assert umv_calls == [] + assert "already at the OUT position" in capsys.readouterr().out + + +def test_losa_out_moves_when_not_out(fake_dev, umv_calls): + optics = make_optics() # currently at "in" positions, not "out" + + optics.losa_out() + + assert umv_calls == [ + (fake_dev.losaz, -1.0), + (fake_dev.losay, 1.0), + ]