feat(flomni,lamni): hard-warn if beam filters are not out before scans

Adds a shared filters_out_of_beam()/warn_and_confirm() helper
(OMNY_shared/filter_check.py) that checks the four cSAXS exposure-box
filter axes (filter_array_1_x..4_x) against their "out" position, reusing
cSAXSFilterTransmission's own position table.

Wired into flomni.py and lamni.py at every point a scan can start:
tomo_scan() and tomo_alignment_scan() (once per run, same interactive/
queued warn-and-confirm convention as the existing X-ray-eye/optics
check), and tomo_scan_projection() (every direct/standalone call, via a
new _internal flag on LamNI's method mirroring flomni's existing one;
internal per-angle loop calls skip the redundant re-check). Since
tomo_queue_execute() already calls tomo_scan(interactive=False) once per
queued job, queued runs get exactly one warn-and-proceed check at queue
start, not one per projection.

DataDrivenLamNI.tomo_scan() (extra_tomo.py) fully overrides the base
method, so it gets its own copy of the same top-of-scan check.

OMNY intentionally not touched here - left as follow-up on the user's
separate OMNY branch, see the OMNY AI_docs TODO in the next commit.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018CUx5yV1iCBhL5sFEHEoPe

csaxs_bec/bec_ipython_client/plugins/LamNI/extra_tomo.py

@@ -18,6 +18,11 @@ import numpy as np
 from bec_lib import bec_logger
 from bec_lib.alarm_handler import AlarmBase
 
+from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.filter_check import (
+    filters_out_of_beam,
+    warn_and_confirm,
+)
+
 from .lamni import LamNI
 
 logger = bec_logger.logger
@@ -56,7 +61,7 @@ class MagLamNI(LamNI):
             lamni_at_each_angle(self, angle)
             return
 
-        self.tomo_scan_projection(angle)
+        self.tomo_scan_projection(angle, _internal=True)
         self.tomo_reconstruct()
 
 
@@ -87,6 +92,14 @@ class DataDrivenLamNI(LamNI):
         """
         bec = builtins.__dict__.get("bec")
         scans = builtins.__dict__.get("scans")
+        dev = builtins.__dict__.get("dev")
+
+        all_out, offending = filters_out_of_beam(dev)
+        if not all_out:
+            warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+            if not warn_and_confirm(self, warning, interactive=True, force=False):
+                print("Aborting tomo scan.")
+                return
 
         bec.builtin_actors.scan_interlock.trigger_setting = "restart_scan"
         bec.builtin_actors.scan_interlock.enabled = True

csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py

@@ -14,6 +14,10 @@ from typeguard import typechecked
 
 from csaxs_bec.bec_ipython_client.plugins.LamNI.gui_tools import LamniGuiTools
 from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni_alignment_mixin import LamNIAlignmentMixin
+from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.filter_check import (
+    filters_out_of_beam,
+    warn_and_confirm,
+)
 from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.omny_general_tools import (
     OMNYTools,
     PtychoReconstructor,
@@ -1159,9 +1163,33 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
     # Scan projection
     # ------------------------------------------------------------------
 
-    def tomo_scan_projection(self, angle: float):
+    def tomo_scan_projection(self, angle: float, _internal: bool = False):
+        """Acquire one fermat-scan projection at `angle`.
+
+        _internal=True is passed by callers that already sit inside an
+        already-checked flow (tomo_scan()'s/tomo_alignment_scan()'s own
+        per-angle loop, via _at_each_angle) so the filter-out-of-beam check
+        below only runs once per tomogram/alignment run instead of once per
+        projection. Direct/standalone calls (the default) are checked every
+        time, mirroring Flomni.tomo_scan_projection()'s _internal convention.
+        """
         scans = builtins.__dict__.get("scans")
 
+        if not _internal:
+            dev = builtins.__dict__.get("dev")
+            all_out, offending = filters_out_of_beam(dev)
+            if not all_out:
+                warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+                if not warn_and_confirm(self, warning, interactive=True, force=False):
+                    # Raise rather than return: this can run inside a
+                    # per-angle loop (directly, or via a custom
+                    # at_each_angle hook), and a bare return would leave the
+                    # projection silently missing / the job stuck running.
+                    raise LamNIError(
+                        "tomo_scan_projection: declined to proceed with filter(s) "
+                        f"in the beam ({', '.join(offending)})."
+                    )
+
         additional_correction = self.compute_additional_correction(angle)
         additional_correction_2 = self.compute_additional_correction_2(angle)
         correction_xeye_mu = self.lamni_compute_additional_correction_xeye_mu(angle)
@@ -1262,6 +1290,13 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
         bec = builtins.__dict__.get("bec")
         dev = builtins.__dict__.get("dev")
 
+        all_out, offending = filters_out_of_beam(dev)
+        if not all_out:
+            warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+            if not warn_and_confirm(self, warning, interactive=True, force=force):
+                print("Aborting alignment scan.")
+                return
+
         self.leye_out()
 
         self.write_alignment_scan_numbers(bec.queue.next_scan_number)
@@ -1281,7 +1316,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
             while not successful:
                 try:
                     start_scan_number = bec.queue.next_scan_number
-                    self.tomo_scan_projection(angle)
+                    self.tomo_scan_projection(angle, _internal=True)
                 except AlarmBase as exc:
                     if exc.alarm_type == "TimeoutError":
                         bec.queue.request_queue_reset()
@@ -1333,7 +1368,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
             hook(self, angle)
             return
 
-        self.tomo_scan_projection(angle)
+        self.tomo_scan_projection(angle, _internal=True)
         self.tomo_reconstruct()
 
     # ------------------------------------------------------------------
@@ -1667,6 +1702,14 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
                 )
                 time.sleep(10)
 
+        dev = builtins.__dict__.get("dev")
+        all_out, offending = filters_out_of_beam(dev)
+        if not all_out:
+            warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+            if not warn_and_confirm(self, warning, interactive=interactive, force=force):
+                print("Aborting tomo scan.")
+                return
+
         self.lamnigui_show_progress()
 
         bec = builtins.__dict__.get("bec")

csaxs_bec/bec_ipython_client/plugins/OMNY_shared/filter_check.py

@@ -0,0 +1,46 @@
+import time
+
+from csaxs_bec.bec_ipython_client.plugins.cSAXS.filter_transmission import cSAXSFilterTransmission
+
+
+def filters_out_of_beam(dev, tol: float = 0.1) -> tuple[bool, list[str]]:
+    """Check whether all cSAXS exposure-box filters are retracted.
+
+    Reuses cSAXSFilterTransmission's own axis list/position table (index 0 of
+    each row is the "out" position) so there is one source of truth for what
+    "out" means, matching csaxs.fil_trans()/_fil_trans_report()'s own check.
+    """
+    offending = []
+    for axis_name, positions in zip(
+        cSAXSFilterTransmission._AXES, cSAXSFilterTransmission._POSITIONS_USER
+    ):
+        axis_obj = getattr(dev, axis_name, None)
+        out_position = positions[0]
+        if axis_obj is None or out_position is None:
+            continue
+        try:
+            rb = float(axis_obj.readback.get())
+        except Exception:
+            continue
+        if abs(rb - out_position) > tol:
+            offending.append(axis_name)
+    return not offending, offending
+
+
+def warn_and_confirm(setup, warning: str, interactive: bool = True, force: bool = False) -> bool:
+    """Eye/optics-check-style warn-and-confirm gate, reusable across plugins.
+
+    Returns True if the caller should proceed. Interactive: red warning + a
+    "Continue anyway?" prompt defaulting to no. Non-interactive (queued/
+    unattended): warns, waits 10s, and always proceeds -- matches
+    LamNI.tomo_scan()'s existing eye/optics gate so queued jobs never hang on
+    input().
+    """
+    if force:
+        return True
+    setup.OMNYTools.printredbold(f"WARNING: {warning}")
+    if interactive:
+        return setup.OMNYTools.yesno("Continue anyway?", "n")
+    setup.OMNYTools.printredbold("Proceeding automatically in 10 s (unattended/queued run)...")
+    time.sleep(10)
+    return True

csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py

@@ -19,6 +19,10 @@ from csaxs_bec.bec_ipython_client.plugins.cSAXS import cSAXSBeamlineChecks
 from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_optics_mixin import FlomniOpticsMixin
 from csaxs_bec.bec_ipython_client.plugins.flomni.gui_tools import flomniGuiTools
 from csaxs_bec.bec_ipython_client.plugins.flomni.x_ray_eye_align import XrayEyeAlign
+from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.filter_check import (
+    filters_out_of_beam,
+    warn_and_confirm,
+)
 from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.tomo_queue_mixin import (
     TomoQueueMixin,
     _GlobalVarParam,
@@ -1787,7 +1791,7 @@ class Flomni(
             upload_url="https://omny.psi.ch/upload.php",
             local_port=8080,
         )
-        self._webpage_gen.start()
+        #self._webpage_gen.start()
 
         self.OMNYTools = OMNYTools(self.client)
         self.reconstructor = PtychoReconstructor(self.ptycho_reconstruct_foldername)
@@ -2197,6 +2201,13 @@ class Flomni(
         dev = builtins.__dict__.get("dev")
         bec = builtins.__dict__.get("bec")
 
+        all_out, offending = filters_out_of_beam(dev)
+        if not all_out:
+            warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+            if not warn_and_confirm(self, warning, interactive=True, force=False):
+                print("Aborting tomo alignment scan.")
+                return
+
         # Only run the full eye-out / optics-in transition if we are not
         # already in measurement condition with feedback running. That
         # transition disables and re-enables-with-reset the rt feedback, which
@@ -2637,8 +2648,10 @@ class Flomni(
         Args:
             interactive: accepted for signature compatibility with
                 LamNI.tomo_scan() (tomo_queue_execute() calls both the same
-                way) -- unused here, FlOMNI has no fine-alignment gate.
+                way). FlOMNI has no fine-alignment gate, but it is used by
+                the filter-out-of-beam check below.
         """
+        dev = builtins.__dict__.get("dev")
 
         if not self._check_eye_out_and_optics_in():
             print(
@@ -2650,6 +2663,13 @@ class Flomni(
                 print("Stopping.")
                 return
 
+        all_out, offending = filters_out_of_beam(dev)
+        if not all_out:
+            warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+            if not warn_and_confirm(self, warning, interactive=interactive, force=False):
+                print("Aborting tomo scan.")
+                return
+
         self.flomnigui_show_progress()
 
         bec = builtins.__dict__.get("bec")
@@ -3343,6 +3363,21 @@ class Flomni(
         _at_each_angle, and tomo_alignment_scan, which needs real ptycho
         data) pass _internal=True to skip the prompt.
         """
+        if not _internal:
+            dev = builtins.__dict__.get("dev")
+            all_out, offending = filters_out_of_beam(dev)
+            if not all_out:
+                warning = f"Not all filters are out of the beam: {', '.join(offending)}."
+                if not warn_and_confirm(self, warning, interactive=True, force=False):
+                    # Raise rather than return, for the same reason as the
+                    # fermat-vs-single-point decline below: this can run
+                    # inside a per-angle loop, and a bare return would leave
+                    # the projection silently missing / the job stuck.
+                    raise FlomniError(
+                        "tomo_scan_projection: declined to proceed with filter(s) "
+                        f"in the beam ({', '.join(offending)})."
+                    )
+
         if not _internal and self.single_point_instead_of_fermat_scan:
             print(
                 "\x1b[93mWarning: single_point_instead_of_fermat_scan is set, but"