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 <noreply@anthropic.com>
This commit is contained in:
x01dc
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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),
|
||||
]
|
||||
Reference in New Issue
Block a user