diff --git a/csaxs_bec/bec_ipython_client/plugins/omny/omny_optics_mixin.py b/csaxs_bec/bec_ipython_client/plugins/omny/omny_optics_mixin.py index 0023308e..3faf811f 100644 --- a/csaxs_bec/bec_ipython_client/plugins/omny/omny_optics_mixin.py +++ b/csaxs_bec/bec_ipython_client/plugins/omny/omny_optics_mixin.py @@ -129,16 +129,44 @@ class OMNYOpticsMixin: if np.fabs(omny_oosa_currentx + 2) > 0.1: raise OMNYError("oosax did not reach target position. Not moving in z.\n") + def _oosa_is_at(self, x_pos, y_pos, z_pos, tol=0.05) -> bool: + """True if oosax/oosay/oosaz are all within tol of the given position.""" + dev.oosaz.read(cached=False) + return ( + np.fabs(dev.oosax.get().readback - x_pos) < tol + and np.fabs(dev.oosay.get().readback - y_pos) < tol + and np.fabs(dev.oosaz.get().readback - z_pos) < tol + ) + def oosa_in(self): + # Compute the target *before* touching anything: _oosa_check_y() below + # unconditionally parks oosay at 0.9, which is not the "in" position for + # either mode, so the "already there" check must happen first or it + # would never see the real (pre-park) oosay reading. + if self.near_field == False: + mode, x_in_pos, y_in_pos, z_in_pos = ( + "far-field", + self._get_user_param_safe("oosax", "far_field_in"), + self._get_user_param_safe("oosay", "far_field_in"), + self._get_user_param_safe("oosaz", "far_field_in"), + ) + else: + mode, x_in_pos, y_in_pos, z_in_pos = ( + "near-field", + self._get_user_param_safe("oosax", "near_field_in"), + self._get_user_param_safe("oosay", "near_field_in"), + self._get_user_param_safe("oosaz", "near_field_in"), + ) + if self._oosa_is_at(x_in_pos, y_in_pos, z_in_pos): + self.OMNYTools.printgreen(f"OSA is already at the IN position ({mode}).") + return + self._oosa_check_y() dev.oshield.read(cached=False) omny_oshield_current = dev.oshield.get().readback if omny_oshield_current < 15: self._oshield_ST_close() if self.near_field == False: - x_in_pos = self._get_user_param_safe("oosax", "far_field_in") - y_in_pos = self._get_user_param_safe("oosay", "far_field_in") - z_in_pos = self._get_user_param_safe("oosaz", "far_field_in") print("OSA movement in far-field mode.") dev.oosaz.read(cached=False) omny_oosa_currentz = dev.oosaz.get().readback @@ -154,9 +182,6 @@ class OMNYOpticsMixin: # umv oosay 0.386015 if self.near_field == True: - x_in_pos = self._get_user_param_safe("oosax", "near_field_in") - y_in_pos = self._get_user_param_safe("oosay", "near_field_in") - z_in_pos = self._get_user_param_safe("oosaz", "near_field_in") print("OSA movement in near-field mode.") dev.oosaz.read(cached=False) omny_oosa_currentz = dev.oosaz.get().readback @@ -178,14 +203,26 @@ class OMNYOpticsMixin: # rt_feedback_enable def oosa_out(self): + # oosa_out() doesn't move oosay, so "already out" only depends on x/z; + # read it before _oosa_check_y() parks oosay at 0.9 below (same reason + # as in oosa_in()). + omny_oosay_current = dev.oosay.get().readback + if self.near_field == False: + mode, z_out_pos = "far-field", 6.5 + else: + mode, z_out_pos = "near-field", -0.45 + if self._oosa_is_at(-2, omny_oosay_current, z_out_pos): + self.OMNYTools.printgreen(f"OSA is already at the OUT position ({mode}).") + return + self._oosa_check_y() dev.oshield.read(cached=False) omny_oshield_current = dev.oshield.get().readback if omny_oshield_current < 15: self._oshield_ST_close() - omny_oosaz_current = dev.oosaz.get().readback if self.near_field == False: print("OSA movement in far-field mode.") + omny_oosaz_current = dev.oosaz.get().readback if omny_oosaz_current < 6.4: self._oosa_to_move_corridor() dev.oosaz.limits = [6.4, 6.6] @@ -193,6 +230,7 @@ class OMNYOpticsMixin: umv(dev.oosax, -2) if self.near_field == True: print("OSA movement in near-field mode.") + omny_oosaz_current = dev.oosaz.get().readback if omny_oosaz_current > 0: self._oosa_to_move_corridor() dev.oosaz.limits = [-0.4, -0.6] @@ -217,18 +255,28 @@ class OMNYOpticsMixin: self.OMNYTools.printgreen("OSA is out of shield.") def ofzp_out(self): + y_out_pos = self._get_user_param_safe("ofzpy", "out") + needs_move = np.fabs(dev.ofzpy.get().readback - y_out_pos) > 0.02 + if not needs_move: + self.OMNYTools.printgreen("FZP is already at the out position") + return if "rtx" in dev and dev.rtx.enabled: dev.rtx.controller.feedback_disable() - y_out_pos = self._get_user_param_safe("ofzpy", "out") - if np.fabs(dev.ofzpy.get().readback - y_out_pos) > 0.02: - umv(dev.ofzpy, y_out_pos) + umv(dev.ofzpy, y_out_pos) self.OMNYTools.printgreen("FZP at out position") def ofzp_in(self): - if "rtx" in dev and dev.rtx.enabled: - dev.rtx.controller.feedback_disable() x_in_pos = self._get_user_param_safe("ofzpx", "in") y_in_pos = self._get_user_param_safe("ofzpy", "in") + needs_move = ( + np.fabs(dev.ofzpy.get().readback - y_in_pos) > 0.02 + or np.fabs(dev.ofzpx.get().readback - x_in_pos) > 0.02 + ) + if not needs_move: + self.OMNYTools.printgreen("FZP is already at the in position") + return + if "rtx" in dev and dev.rtx.enabled: + dev.rtx.controller.feedback_disable() if np.fabs(dev.ofzpy.get().readback - y_in_pos) > 0.02: umv(dev.ofzpy, y_in_pos) diff --git a/tests/tests_bec_ipython_client/test_omny_optics_skip_if_in_position.py b/tests/tests_bec_ipython_client/test_omny_optics_skip_if_in_position.py new file mode 100644 index 00000000..bbbc5700 --- /dev/null +++ b/tests/tests_bec_ipython_client/test_omny_optics_skip_if_in_position.py @@ -0,0 +1,204 @@ +"""Tests for omny's "skip if already in position" optics optimization for +oosa_in()/oosa_out()/ofzp_in()/ofzp_out(), ported from flomni's +fosa_in()/ffzp_in() (same optimization already done for lamni, see +test_lamni_optics_skip_if_in_position.py). + +Methods under test (omny_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.omny.omny_optics_mixin as optics_mixin_module +from csaxs_bec.bec_ipython_client.plugins.omny.omny_optics_mixin import OMNYOpticsMixin + +# far-field in: oosax=3.022, oosay=0.4122, oosaz=6.5 (out) +# near-field in: oosax=3.2044, oosay=0.531, oosaz=-0.4452 +FAR_IN = {"oosax": 3.022, "oosay": 0.4122, "oosaz": 6.5} +NEAR_IN = {"oosax": 3.2044, "oosay": 0.531, "oosaz": -0.4452} + + +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 + + def read(self, cached=True): + return self.value + + def get(self): + return types.SimpleNamespace(readback=self.value) + + +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, oosay=0.9, oshield=20.0, osamy=0.0): + self.oosax = FakeAxis(0.0, in_pos=FAR_IN["oosax"]) + self.oosax.user_parameter = { + "far_field_in": FAR_IN["oosax"], + "near_field_in": NEAR_IN["oosax"], + } + self.oosay = FakeAxis(oosay) + self.oosay.user_parameter = { + "far_field_in": FAR_IN["oosay"], + "near_field_in": NEAR_IN["oosay"], + } + self.oosaz = FakeAxis(0.0) + self.oosaz.limits = None + self.oosaz.user_parameter = { + "far_field_in": FAR_IN["oosaz"], + "near_field_in": NEAR_IN["oosaz"], + } + self.oshield = FakeAxis(oshield) + self.osamy = FakeAxis(osamy) + self.ofzpx = FakeAxis(-0.5, in_pos=-0.5) + self.ofzpy = FakeAxis(0.3, in_pos=0.3, out_pos=0.0) + self.rtx = types.SimpleNamespace(enabled=True, controller=FakeController()) + + def __contains__(self, name): + return hasattr(self, name) + + def __getitem__(self, name): + return getattr(self, name) + + +@pytest.fixture +def fake_dev(monkeypatch): + # omny_optics_mixin.py has no module-level `dev`/`umv` names of its own -- + # it resolves them via the builtins fallback at call time, so the module + # attribute must be force-created (raising=False) rather than assumed to + # pre-exist. + dev = FakeDev() + monkeypatch.setattr(optics_mixin_module, "dev", dev, raising=False) + return dev + + +@pytest.fixture +def umv_calls(monkeypatch): + calls = [] + monkeypatch.setattr(optics_mixin_module, "umv", lambda *args: calls.append(args), raising=False) + return calls + + +def make_optics(near_field=False): + optics = OMNYOpticsMixin() + optics.near_field = near_field + optics.OMNYTools = types.SimpleNamespace(printgreen=print) + return optics + + +def test_oosa_in_skips_move_when_already_in_position_far_field(fake_dev, umv_calls, capsys): + fake_dev.oosax.value = FAR_IN["oosax"] + fake_dev.oosay.value = FAR_IN["oosay"] + fake_dev.oosaz.value = FAR_IN["oosaz"] + optics = make_optics(near_field=False) + + optics.oosa_in() + + assert umv_calls == [] + assert "already at the IN position" in capsys.readouterr().out + # _oosa_check_y()/shield-close must not run either -- oosay untouched + assert fake_dev.oosay.value == FAR_IN["oosay"] + + +def test_oosa_in_moves_when_out_of_position(fake_dev, umv_calls): + fake_dev.oosax.value = -2.0 # not at far-field "in" + optics = make_optics(near_field=False) + + optics.oosa_in() + + assert (fake_dev.oosaz, FAR_IN["oosaz"]) in umv_calls + assert (fake_dev.oosax, FAR_IN["oosax"]) in umv_calls + assert (fake_dev.oosay, FAR_IN["oosay"]) in umv_calls + + +def test_oosa_in_skips_move_when_already_in_position_near_field(fake_dev, umv_calls, capsys): + fake_dev.oosax.value = NEAR_IN["oosax"] + fake_dev.oosay.value = NEAR_IN["oosay"] + fake_dev.oosaz.value = NEAR_IN["oosaz"] + optics = make_optics(near_field=True) + + optics.oosa_in() + + assert umv_calls == [] + assert "already at the IN position" in capsys.readouterr().out + + +def test_oosa_out_skips_move_when_already_out(fake_dev, umv_calls, capsys): + fake_dev.oosax.value = -2.0 + fake_dev.oosaz.value = 6.5 + optics = make_optics(near_field=False) + + optics.oosa_out() + + assert umv_calls == [] + assert "already at the OUT position" in capsys.readouterr().out + + +def test_oosa_out_moves_when_not_out(fake_dev, umv_calls): + fake_dev.oosax.value = FAR_IN["oosax"] + fake_dev.oosaz.value = FAR_IN["oosaz"] + optics = make_optics(near_field=False) + + optics.oosa_out() + + assert (fake_dev.oosaz, 6.5) in umv_calls + assert (fake_dev.oosax, -2) in umv_calls + + +def test_ofzp_in_skips_move_and_feedback_disable_when_already_in(fake_dev, umv_calls, capsys): + optics = make_optics() + + optics.ofzp_in() + + assert umv_calls == [] + assert fake_dev.rtx.controller.calls == [] + assert "already at the in position" in capsys.readouterr().out + + +def test_ofzp_in_moves_and_disables_feedback_when_out_of_position(fake_dev, umv_calls): + fake_dev.ofzpx.value = -1.0 # not at "in" (-0.5) + optics = make_optics() + + optics.ofzp_in() + + assert (fake_dev.ofzpx, -0.5) in umv_calls + assert fake_dev.rtx.controller.calls == ["disable"] + + +def test_ofzp_out_skips_move_and_feedback_disable_when_already_out(fake_dev, umv_calls, capsys): + fake_dev.ofzpy.value = 0.0 # matches "out" + optics = make_optics() + + optics.ofzp_out() + + assert umv_calls == [] + assert fake_dev.rtx.controller.calls == [] + assert "already at the out position" in capsys.readouterr().out + + +def test_ofzp_out_moves_and_disables_feedback_when_not_out(fake_dev, umv_calls): + optics = make_optics() # ofzpy at 0.3, not "out" (0.0) + + optics.ofzp_out() + + assert (fake_dev.ofzpy, 0.0) in umv_calls + assert fake_dev.rtx.controller.calls == ["disable"]