csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py

@@ -2139,8 +2139,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
     def _get_val(msg: str, default_value, data_type):
         return data_type(input(f"{msg} ({default_value}): ") or default_value)
 
-    @staticmethod
-    def _confirm_sequence_override(warning: str, force: bool) -> bool:
+    def _confirm_sequence_override(self, warning: str, force: bool) -> bool:
         """Print *warning* and ask whether to proceed anyway.
 
         Used by the alignment-sequence gates (xrayeye_alignment_start(),
@@ -2161,8 +2160,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
         if force:
             return True
         print(warning)
-        answer = input("Continue anyway? [y/N]: ").strip().lower()
-        return answer in ("y", "yes")
+        return self.OMNYTools.yesno("Continue anyway?", "n")
 
     # ------------------------------------------------------------------
     # PDF report

csaxs_bec/bec_ipython_client/plugins/LamNI/lamni_optics_mixin.py

@@ -227,12 +227,8 @@ class LamNIOpticsMixin:
         bec.queue.next_dataset_number += 1
         umv(dev.lsamrot, 0)
         #umv(dev.dttrz, 6419.677, dev.fttrz, 2959.979)
-        while True:
-            moved_out = (input("Did the flight tube move out? (Y/n)") or "y").lower()
-            if moved_out == "y":
-                break
-            if moved_out == "n":
-                return
+        if not self.OMNYTools.yesno("Did the flight tube move out?", "y"):
+            return
         leyex_in = self._get_user_param_safe("leyex", "in")
         leyey_in = self._get_user_param_safe("leyey", "in")
         umv(dev.leyex, leyex_in, dev.leyey, leyey_in)

csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py

@@ -544,8 +544,7 @@ class XrayEyeAlign:
             self.gui.set_dap_params_forwarding(False)
 
         if keep_shutter_open:
-            answer = input("Close the shutter now? [Y/n]: ").strip().lower()
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno("Close the shutter now?", "y"):
                 dev.fsh.fshclose()
                 self.gui.on_live_view_enabled(False)
                 print("Shutter closed.")
@@ -961,12 +960,9 @@ class XrayEyeAlign:
             # wrap) with live view running.
             self._live_sweep([0])
             self.update_frame(keep_shutter_open)
-            answer = (
-                input("Alignment acceptable -- stop here? [Y/n] (n = run another iteration): ")
-                .strip()
-                .lower()
-            )
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno(
+                "Alignment acceptable -- stop here? (n = run another iteration)", "y"
+            ):
                 print(
                     f"[rotation-center][smear] operator accepted alignment after "
                     f"{iteration} iteration(s)"
@@ -985,15 +981,11 @@ class XrayEyeAlign:
         )
 
         if apply:
-            answer = (
-                input(
-                    f"Update lsamx/lsamy center user parameters to "
-                    f"({new_lsamx:.4f}, {new_lsamy:.4f})? [Y/n]: "
-                )
-                .strip()
-                .lower()
-            )
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno(
+                f"Update lsamx/lsamy center user parameters to "
+                f"({new_lsamx:.4f}, {new_lsamy:.4f})?",
+                "y",
+            ):
                 dev.lsamx.update_user_parameter({"center": float(new_lsamx)})
                 dev.lsamy.update_user_parameter({"center": float(new_lsamy)})
                 self._mark_center_found_and_invalidate_downstream()
@@ -1009,8 +1001,7 @@ class XrayEyeAlign:
                 )
 
         if keep_shutter_open:
-            answer = input("Close the shutter now? [Y/n]: ").strip().lower()
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno("Close the shutter now?", "y"):
                 dev.fsh.fshclose()
                 self.gui.on_live_view_enabled(False)
                 print("Shutter closed.")
@@ -1293,14 +1284,9 @@ class XrayEyeAlign:
                 self.send_message("Verifying alignment...")
                 self._live_sweep([45, 0])
                 self.update_frame(keep_shutter_open)
-                answer = (
-                    input(
-                        "Alignment acceptable -- stop here? [Y/n] (n = run another iteration): "
-                    )
-                    .strip()
-                    .lower()
-                )
-                if answer in ("", "y", "yes"):
+                if self.lamni.OMNYTools.yesno(
+                    "Alignment acceptable -- stop here? (n = run another iteration)", "y"
+                ):
                     print(
                         f"[rotation-center] operator accepted alignment after {iteration} iteration(s)"
                     )
@@ -1318,15 +1304,11 @@ class XrayEyeAlign:
         )
 
         if apply:
-            answer = (
-                input(
-                    f"Update lsamx/lsamy center user parameters to "
-                    f"({new_lsamx:.4f}, {new_lsamy:.4f})? [Y/n]: "
-                )
-                .strip()
-                .lower()
-            )
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno(
+                f"Update lsamx/lsamy center user parameters to "
+                f"({new_lsamx:.4f}, {new_lsamy:.4f})?",
+                "y",
+            ):
                 dev.lsamx.update_user_parameter({"center": float(new_lsamx)})
                 dev.lsamy.update_user_parameter({"center": float(new_lsamy)})
                 self._mark_center_found_and_invalidate_downstream()
@@ -1342,8 +1324,7 @@ class XrayEyeAlign:
                 )
 
         if keep_shutter_open:
-            answer = input("Close the shutter now? [Y/n]: ").strip().lower()
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno("Close the shutter now?", "y"):
                 dev.fsh.fshclose()
                 self.gui.on_live_view_enabled(False)
                 print("Shutter closed.")

csaxs_bec/bec_ipython_client/plugins/omny/omny_optics_mixin.py

@@ -28,10 +28,9 @@ class OMNYOpticsMixin:
 
         self.align.update_frame()
 
-        user_input = input(
+        if self.OMNYTools.yesno(
             "Is the direct beam gone on the xray eye? Do you see the cone of the FZP?"
-        )
-        if user_input == "y":
+        ):
             printf("Next oeye_out...\n")
         else:
             raise OMNYError("Failed to properly move in the Xray optics")

csaxs_bec/bec_ipython_client/plugins/omny/x_ray_eye_align.py

@@ -355,8 +355,7 @@ class XrayEyeAlign:
         self.lamni.align.read_xray_eye_correction_from_gui()
 
         if keep_shutter_open:
-            answer = input("Close the shutter now? [Y/n]: ").strip().lower()
-            if answer in ("", "y", "yes"):
+            if self.lamni.OMNYTools.yesno("Close the shutter now?", "y"):
                 dev.omnyfsh.fshclose()
                 self.gui.on_live_view_enabled(False)
                 print("Shutter closed.")

tests/tests_bec_ipython_client/test_x_ray_eye_align.py

@@ -16,6 +16,10 @@ from csaxs_bec.bec_ipython_client.plugins.LamNI.x_ray_eye_align import XrayEyeAl
 # pylint: disable=protected-access
 
 XRAY_EYE_ALIGN = "csaxs_bec.bec_ipython_client.plugins.LamNI.x_ray_eye_align"
+# find_rotation_center()'s y/n prompts go through OMNYTools.yesno(), which
+# calls the real input() from this module -- not from XRAY_EYE_ALIGN -- so
+# that's what needs patching to keep these tests from blocking on stdin.
+OMNY_GENERAL_TOOLS = "csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools"
 
 
 class RTControllerMock:
@@ -234,7 +238,7 @@ def test_extended_calibration_accumulates_shift_across_iterations(bec_client_moc
     with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
         with mock.patch(f"{XRAY_EYE_ALIGN}.umv"):
             with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
-                with mock.patch(f"{XRAY_EYE_ALIGN}.input", side_effect=["n", "y"]):
+                with mock.patch(f"{OMNY_GENERAL_TOOLS}.input", side_effect=["n", "y"]):
                     align.find_rotation_center(sample_type="extended", apply=False)
 
     calls = align.scans.lamni_move_to_scan_center.call_args_list
@@ -309,7 +313,7 @@ def test_find_rotation_center_isolated_computes_midpoint(bec_client_mock):
     with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
         with mock.patch(f"{XRAY_EYE_ALIGN}.umv"):
             with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
-                with mock.patch(f"{XRAY_EYE_ALIGN}.input", return_value="y"):
+                with mock.patch(f"{OMNY_GENERAL_TOOLS}.input", return_value="y"):
                     result = align.find_rotation_center(sample_type="isolated")
 
     # midpoint of (0.3,0.3) and (0.5,0.7) is (0.4, 0.5) ->
@@ -337,7 +341,7 @@ def test_find_rotation_center_isolated_declines_apply(bec_client_mock):
     with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock):
         with mock.patch(f"{XRAY_EYE_ALIGN}.umv"):
             with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
-                with mock.patch(f"{XRAY_EYE_ALIGN}.input", return_value="n"):
+                with mock.patch(f"{OMNY_GENERAL_TOOLS}.input", return_value="n"):
                     align.find_rotation_center(sample_type="isolated")
 
     dev_mock.lsamx.update_user_parameter.assert_not_called()
@@ -364,7 +368,7 @@ def test_find_rotation_center_extended_iterates_until_happy(bec_client_mock):
             with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
                 # first iteration: "not happy" -> second iteration: "happy",
                 # then accept the final apply prompt
-                with mock.patch(f"{XRAY_EYE_ALIGN}.input", side_effect=["n", "y", "y"]):
+                with mock.patch(f"{OMNY_GENERAL_TOOLS}.input", side_effect=["n", "y", "y"]):
                     align.find_rotation_center(sample_type="extended")
 
     # one lamni_move_to_scan_center pair per iteration -> 2 iterations = 4 calls
@@ -546,7 +550,7 @@ def test_find_rotation_center_smear_experimental_switches_live_view_signal(bec_c
         with mock.patch(f"{XRAY_EYE_ALIGN}.umv"):
             with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report):
                 with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
-                    with mock.patch(f"{XRAY_EYE_ALIGN}.input", side_effect=["y", "y"]):
+                    with mock.patch(f"{OMNY_GENERAL_TOOLS}.input", side_effect=["y", "y"]):
                         with mock.patch.object(align, "_save_smear_calibration_data"):
                             align.find_rotation_center_smear_experimental()
 
@@ -572,7 +576,7 @@ def test_find_rotation_center_smear_experimental_saves_composites_and_clicks(bec
         with mock.patch(f"{XRAY_EYE_ALIGN}.umv"):
             with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report):
                 with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"):
-                    with mock.patch(f"{XRAY_EYE_ALIGN}.input", side_effect=["y", "y"]):
+                    with mock.patch(f"{OMNY_GENERAL_TOOLS}.input", side_effect=["y", "y"]):
                         with mock.patch.object(align, "_save_smear_calibration_data") as save_mock:
                             align.find_rotation_center_smear_experimental(sweep_deg=45.0)