bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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
This commit is contained in:
x12sa
2026-09-10 15:40:52 +02:00
co-authored by Claude Sonnet 5
parent 36e19adde4
commit 7d64b49918
4 changed files with 143 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/LamNI/extra_tomo.py
+14 -1
View File
@@ -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
+46 -3
View File
@@ -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
+46
View File
@@ -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
+37 -2
View File
@@ -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"