Lamni minor fixes #290

Merged
holler merged 8 commits from lamni_minor_fixes into main 2026-07-31 13:43:48 +02:00
19 changed files with 592 additions and 81 deletions
@@ -1650,6 +1650,22 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
check instead prints a bold warning, waits 10s, and always
proceeds -- it never aborts or raises.
"""
if not self._check_eye_out_and_optics_in():
warning = (
"The setup is not in measurement configuration -- the X-ray "
"eye might be IN or the optics (FZP/OSA) OUT."
)
if interactive:
if not self._confirm_sequence_override(warning, force):
print("Aborting tomo scan.")
return
else:
self.OMNYTools.printredbold(f"WARNING: {warning}")
self.OMNYTools.printredbold(
"Proceeding automatically in 10 s (unattended/queued run)..."
)
time.sleep(10)
self.lamnigui_show_progress()
bec = builtins.__dict__.get("bec")
@@ -1740,6 +1756,12 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
self.progress["tomo_type"] = "Golden ratio tomography"
previous_subtomo_number = -1
ii = 0 if projection_number is None else projection_number
# Suppresses the 0-deg shot only for the first sub-tomogram
# transition detected below -- on a resume that lands on an
# odd sub-tomogram past projection 10, that first detected
# "transition" is an artifact of previous_subtomo_number
# starting at -1, not an actual pass through 0 deg.
resuming = projection_number is not None
while True:
angle, subtomo_number = self._golden(
ii, self.golden_ratio_bunch_size, maxangle=360, reverse=True
@@ -1749,9 +1771,11 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
and not resuming
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
resuming = False
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
@@ -1778,6 +1802,12 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
self.progress["tomo_type"] = "Equally spaced, golden ratio starting angles"
previous_subtomo_number = -1
ii = 0 if projection_number is None else projection_number
# Suppresses the 0-deg shot only for the first sub-tomogram
# transition detected below -- on a resume that lands on an
# odd sub-tomogram past projection 10, that first detected
# "transition" is an artifact of previous_subtomo_number
# starting at -1, not an actual pass through 0 deg.
resuming = projection_number is not None
while True:
angle, subtomo_number = self._golden_equally_spaced(
ii, int(360 / self.tomo_angle_stepsize), maxangle=360, reverse=True
@@ -1787,9 +1817,11 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
and not resuming
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
resuming = False
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
@@ -2103,8 +2135,6 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
int,
)
self.sample_name = self._get_val("sample name", self.sample_name, str)
@staticmethod
def _get_val(msg: str, default_value, data_type):
return data_type(input(f"{msg} ({default_value}): ") or default_value)
@@ -312,6 +312,50 @@ class LamNIOpticsMixin:
self.lfzp_in()
self.losa_in()
def _check_eye_out_and_optics_in(self, tol=0.003):
"""True if the setup is in measurement configuration: X-ray eye OUT,
FZP and OSA IN. Mirrors flomni's
``FlomniOpticsMixin._check_eye_out_and_optics_in()``, adapted to
LamNI's axes -- leye_out() only moves leyey (not leyex) to its "out"
position, so the eye-out condition here is checked on leyey.
"""
# --- expected IN positions ---
loptx_in = self._get_user_param_safe("loptx", "in")
lopty_in = self._get_user_param_safe("lopty", "in")
# --- expected OUT condition for the X-ray eye ---
leyey_out = self._get_user_param_safe("leyey", "out")
# --- current positions ---
cx_leyey = dev.leyey.readback.get()
cx_loptx = dev.loptx.readback.get()
cx_lopty = dev.lopty.readback.get()
# --- check eye OUT ---
eye_out = np.isclose(cx_leyey, leyey_out, atol=tol)
# --- check optics IN ---
optics_in = np.isclose(cx_loptx, loptx_in, atol=tol) and np.isclose(
cx_lopty, lopty_in, atol=tol
)
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")
cx_losax = dev.losax.readback.get()
cx_losay = dev.losay.readback.get()
cx_losaz = dev.losaz.readback.get()
osa_in = (
np.isclose(cx_losax, losax_in, atol=tol)
and np.isclose(cx_losay, losay_in, atol=tol)
and np.isclose(cx_losaz, losaz_in, atol=tol)
)
return eye_out and optics_in and osa_in
def loptics_out(self):
"""Move out the LamNI optics."""
if "rtx" in dev and dev.rtx.enabled:
@@ -153,16 +153,20 @@ class XrayEyeAlign:
def _enable_rt_feedback(self):
self.device_manager.devices.rtx.controller.feedback_enable_with_reset()
def _gui_call_with_retry(self, func, *args, retries: int = 8, retry_delay: float = 1.5, **kwargs):
"""Call a bec_widgets GUI RPC method, retrying briefly on "GUI is not alive".
def _gui_call_with_retry(self, func, *args, retries: int = 25, retry_delay: float = 2.0, **kwargs):
"""Call a bec_widgets GUI RPC method, retrying on "GUI is not alive".
The GUI's liveness check is a Redis heartbeat with a 10 s TTL, refreshed
every 200 ms from the same Qt event loop that renders live-view frames.
A long blocking device move (e.g. interferometer feedback reset, a live
rotation sweep) with live view left on can starve that heartbeat past
its TTL even though the GUI process is still alive -- it catches up and
resumes the heartbeat shortly after the block ends, so a short retry
loop recovers without needing any change to bec_widgets itself.
rotation sweep covering up to ~360 deg) with live view left on can
starve that heartbeat past its TTL even though the GUI process is
still alive -- it catches up and resumes the heartbeat shortly after
the block ends, so a retry loop recovers without needing any change to
bec_widgets itself. The default budget (25 retries * 2.0 s = 48 s
total wait) is sized to comfortably outlast a full-range rotation
sweep on slower hardware, where such a move has been observed to take
30+ s.
"""
for attempt in range(retries):
try:
@@ -949,7 +953,13 @@ class XrayEyeAlign:
self._apply_rotation_center_shift(cumulative_shift_x, cumulative_shift_y)
self.send_message("Verifying alignment...")
self._live_sweep([45, 0])
# No 45 deg detour here (unlike the non-smear path below): the
# _smear_sweep above already rotated continuously through the
# full range with the operator watching, so a stop at 45 deg
# would just re-tread ground already covered -- go straight back
# to 0 deg (still a full ~360 deg move since lsamrot doesn't
# 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): ")
@@ -2754,12 +2754,6 @@ function renderTomoQueue(jobs){{
el.innerHTML='<div class="tq-empty">No jobs queued</div>';
return;
}}
// Running jobs are auto-opened once, on first appearance. We remember which
// we have already auto-opened so a manual collapse is not undone on the next
// refresh (previously, collapsing the only open job made the snapshot empty
// and re-triggered the auto-open).
const autoOpened=window.__tqAutoOpened||(window.__tqAutoOpened=new Set());
const presentKeys=new Set();
let html='';
jobs.forEach((job,i)=>{{
const status=job.status||'pending';
@@ -2767,14 +2761,9 @@ function renderTomoQueue(jobs){{
const isCommand=job.kind==='command';
const params=job.params||{{}};
const key=(job.added_at||'')+'|'+label;
presentKeys.add(key);
// Open if the user currently has it open, or it is a running job we have
// not auto-opened before (its first appearance). Auto-open fires once.
let shouldOpen=openKeys.has(key);
if(status==='running' && !autoOpened.has(key)){{
shouldOpen=true;
autoOpened.add(key);
}}
// Open only if the user currently has it open -- never force-open a
// job just because it started running.
const shouldOpen=openKeys.has(key);
// Command jobs have no "params" key -- render their step sequence
// instead of an (otherwise blank) params table.
const paramRows=isCommand?'':buildParamRows(params);
@@ -2799,9 +2788,6 @@ function renderTomoQueue(jobs){{
+'</div>'
+'</details>';
}});
// Drop remembered auto-open keys for jobs that are gone, so the set stays
// small and a job that later reappears is treated as new.
autoOpened.forEach(k=>{{ if(!presentKeys.has(k)) autoOpened.delete(k); }});
el.innerHTML=html;
}}
@@ -2593,6 +2593,12 @@ class Flomni(
ii = 0
else:
ii = projection_number
# Suppresses the 0-deg shot only for the first sub-tomogram
# transition detected below -- on a resume that lands on an
# odd sub-tomogram past projection 10, that first detected
# "transition" is an artifact of previous_subtomo_number
# starting at -1, not an actual pass through 0 deg.
resuming = projection_number is not None
while True:
angle, subtomo_number = self._golden(
ii, self.golden_ratio_bunch_size, 180, 1
@@ -2602,9 +2608,11 @@ class Flomni(
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
and not resuming
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
resuming = False
self.progress["tomo_type"] = "Golden ratio tomography"
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
@@ -2644,6 +2652,12 @@ class Flomni(
ii = 0
else:
ii = projection_number
# Suppresses the 0-deg shot only for the first sub-tomogram
# transition detected below -- on a resume that lands on an
# odd sub-tomogram past projection 10, that first detected
# "transition" is an artifact of previous_subtomo_number
# starting at -1, not an actual pass through 0 deg.
resuming = projection_number is not None
while True:
angle, subtomo_number = self._golden_equally_spaced(
ii, int(180 / self.tomo_angle_stepsize), 180, 1, 0
@@ -2653,9 +2667,11 @@ class Flomni(
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
and not resuming
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
resuming = False
self.progress["tomo_type"] = (
"Equally spaced tomography, golden ratio starting angle"
)
@@ -2839,26 +2839,15 @@ function renderTomoQueue(jobs){{
el.innerHTML='<div class="tq-empty">No jobs queued</div>';
return;
}}
// Running jobs are auto-opened once, on first appearance. We remember which
// we have already auto-opened so a manual collapse is not undone on the next
// refresh (previously, collapsing the only open job made the snapshot empty
// and re-triggered the auto-open).
const autoOpened=window.__tqAutoOpened||(window.__tqAutoOpened=new Set());
const presentKeys=new Set();
let html='';
jobs.forEach((job,i)=>{{
const status=job.status||'pending';
const label=job.label||('Job '+(i+1));
const params=job.params||{{}};
const key=(job.added_at||'')+'|'+label;
presentKeys.add(key);
// Open if the user currently has it open, or it is a running job we have
// not auto-opened before (its first appearance). Auto-open fires once.
let shouldOpen=openKeys.has(key);
if(status==='running' && !autoOpened.has(key)){{
shouldOpen=true;
autoOpened.add(key);
}}
// Open only if the user currently has it open -- never force-open a
// job just because it started running.
const shouldOpen=openKeys.has(key);
const paramRows=buildParamRows(params);
const addedAt=job.added_at
?'Added '+new Date(job.added_at).toLocaleString([],{{month:'short',day:'numeric',hour:'2-digit',minute:'2-digit'}})
@@ -2875,9 +2864,6 @@ function renderTomoQueue(jobs){{
+'</div>'
+'</details>';
}});
// Drop remembered auto-open keys for jobs that are gone, so the set stays
// small and a job that later reappears is treated as new.
autoOpened.forEach(k=>{{ if(!presentKeys.has(k)) autoOpened.delete(k); }});
el.innerHTML=html;
}}
@@ -307,8 +307,6 @@ cam_xeye:
readoutPriority: async
userParameter:
pixel_calibration: 0.0009174311926605505 # mm/pixel (= 0.2 / 218)
############################################################
deviceTags:
- ptycho_lamni
#################### GUI Signals ###########################
+2 -2
View File
@@ -172,8 +172,8 @@ class LamniGalilController(GalilController):
def lgalil_is_air_off_and_orchestra_enabled(self) -> bool:
# TODO: move this to the LamNI-specific controller
rt_not_blocked_by_galil = bool(self.socket_put_and_receive("MG@OUT[9]"))
air_off = bool(self.socket_put_and_receive("MG@OUT[13]"))
rt_not_blocked_by_galil = bool(float(self.socket_put_and_receive("MG@OUT[9]")))
air_off = bool(float(self.socket_put_and_receive("MG@OUT[13]")))
return rt_not_blocked_by_galil and air_off
+5 -1
View File
@@ -371,7 +371,11 @@ class FlomniFermatScan(ScanBase):
angle (float): The target angle for the flomni rotation.
"""
fsamroy_current_setpoint = self.dev.fsamroy.user_setpoint.get()
if angle == fsamroy_current_setpoint:
fsamroy_readback = self.dev.fsamroy.readback.get()
already_at_angle = angle == fsamroy_current_setpoint and np.isclose(
fsamroy_readback, angle, atol=self.dev.fsamroy.tolerance
)
if already_at_angle:
logger.info("No rotation required")
return
+9 -2
View File
@@ -481,14 +481,21 @@ class LamniFermatScan(ScanBase):
def lamni_rotation(self, angle: float):
"""
Rotate LamNI to the specified angle. The rotation is only performed
if the current setpoint of the rotation stage is different from the requested angle.
if the current setpoint of the rotation stage is different from the requested angle,
or the stage did not actually settle into measurement configuration
(e.g. a previous rotation was interrupted before the air bearings clamped
and the piezo stage was released back to interferometer feedback).
Args:
angle (float): Rotation angle in degrees
"""
# get last setpoint (cannot be based on pos get because they will deviate slightly)
lsamrot_current_setpoint = self.dev.lsamrot.user_setpoint.get()
if angle == lsamrot_current_setpoint:
already_at_angle = (
angle == lsamrot_current_setpoint
and self.dev.lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled()
)
if already_at_angle:
logger.info("No rotation required")
return
+5 -1
View File
@@ -347,7 +347,11 @@ class OmnyFermatScan(ScanBase):
angle (float): Rotation angle in degrees.
"""
osamroy_current_setpoint = self.dev.osamroy.user_setpoint.get()
if angle == osamroy_current_setpoint:
osamroy_readback = self.dev.osamroy.readback.get()
already_at_angle = angle == osamroy_current_setpoint and np.isclose(
osamroy_readback, angle, atol=self.dev.osamroy.tolerance
)
if already_at_angle:
logger.info("No rotation required.")
return
logger.info("Rotating to requested angle")
+9 -27
View File
@@ -20,22 +20,20 @@ Mount the new sample. The X-ray eye is already in, but the X-ray optics needs to
The effective position of the axis of rotation shifts with sample thickness or mounting position of the sample along the axis of rotation. The position of the axis of rotation is controlled by user parameters **center** of the **lsamx** and **lsamy** stages.
The recommended way to (re-)measure `center` for a new sample is the GUI-driven procedure below. It replaces the fully manual jog-by-hand steps that used to live here; those are kept further down as a fallback for when the automated procedure or the GUI is unavailable.
The recommended way to (re-)measure center for a new sample is the GUI-driven procedure below. It replaces the fully manual jog-by-hand steps that used to live here; those are kept further down as a fallback for when the automated procedure or the GUI is unavailable.
**Automated rotation-center calibration**
This opens the X-ray eye widget and walks through the calibration interactively. Pick the function based on what's mounted:
- `lamni.xrayeye_rotation_center_calibration_isolated()` — for a sparse/isolated particle. You submit its centre position once at 0° and once at 180°; the midpoint of the two is the rotation axis' projected position (this works regardless of the axis tilt). No further confirmation step.
- `lamni.xrayeye_rotation_center_calibration_extended()` — for a non-isolated/textured sample where the rotation centre can be identified visually. The sample sweeps 0° → 180° → 0° with the live view left open so you can watch for the point that doesn't move, then you submit a single click at 0°. The correction is applied, a short verification sweep (0° → 45° → 0°) runs, and you're asked *"Alignment acceptable — stop here?"* — answer `n` to refine further.
- `lamni.xrayeye_rotation_center_calibration_isolated()` — for a sparse/isolated particle. You submit its centre position once at 0° and once at 180°; the midpoint of the two is the rotation axis' projected position. Pass `keep_shutter_open=True` if it's hard to relocate the sample between steps, matching `xrayeye_alignment_start()` below.
- `lamni.xrayeye_rotation_center_calibration_extended()` — for a non-isolated/textured sample where the rotation centre can be identified visually. The sample sweeps 0° → 180° → 0° with the live view left open so you can watch for the point that doesn't move, then you submit the rotation centre. The correction is applied, a short verification sweep (0° → 45° → 0°) runs, and you're asked *"Alignment acceptable — stop here?"* — answer `n` to run again.
At the end you're shown the computed `lsamx_center`/`lsamy_center` and asked to confirm before they're written via `dev.lsamx.update_user_parameter(...)` / `dev.lsamy.update_user_parameter(...)`.
Pass `keep_shutter_open=True` if it's hard to relocate the sample between steps, matching `xrayeye_alignment_start()` below. Interrupt with Ctrl-C to abort; the crosshair and DAP forwarding are cleaned up automatically.
**Experimental: continuous-rotation "smear" aid**
`lamni.xrayeye_rotation_center_calibration_smear_experimental(sweep_deg=360.0, keep_shutter_open=False)` is an **experimental** alternative to `..._extended()` for the same non-isolated-sample case. Instead of judging the rotation centre from a single instant of live rotation, it rotates `lsamrot` continuously through `sweep_deg` (default a full circle) while accumulating a max-projection ("star-trail") composite from the camera: off-axis features smear into circular arcs, and the common center of curvature of those arcs is the rotation axis — usually an easier target to click than one live frame. The composite builds up live on screen as the sweep progresses (pushed to its own preview channel, so the "camera running" indicator stays steady throughout — see {ref}`the developer notes <developer.lamni_smear_architecture>` for why that matters) and stays frozen once the sweep ends, until you submit your click. `sweep_deg` can be reduced below 360 (even below 180) if a shorter arc already shows enough curvature. There is no automatic circle fitting — you still submit the centre by eye, same click mechanism as `..._extended()`, and the same iterate/verify/apply flow follows. Not yet merged into the production branch; try it from `experimental/rotation_center_smear`.
`lamni.xrayeye_rotation_center_calibration_smear_experimental(sweep_deg=360.0)` is an **experimental** alternative to `..._extended()` for the same non-isolated-sample case. Instead of judging the rotation centre from a single instant of live rotation, it rotates `lsamrot` continuously through `sweep_deg` while accumulating a max-projection ("star-trail") composite from the camera: off-axis features smear into circular arcs, and the common center of curvature of those arcs is the rotation axis — usually an easier target to click than one live frame. The composite builds up live on screen as the sweep progresses and stays frozen once the sweep ends, until you submit your click. `sweep_deg` can be reduced below 360 if a shorter arc already shows enough curvature. There is no automatic circle fitting — you still submit the centre by eye, same click mechanism as `..._extended()`.
**Manual fallback**
@@ -53,7 +51,7 @@ To observe the axis of rotation obtain the position of the Fresnel zone plate on
#### X-ray eye alignment
The BEC GUI is used to obtain a coarse sample alignment. Start the alignment process (which clears any previous alignment) by
The BEC GUI is used to obtain a coarse sample alignment. Start the alignment process by
`lamni.xrayeye_alignment_start()`
This opens the X-ray eye widget automatically. The procedure collects the sample centre position at 8 angles (0°–315° in 45° steps, full 360° rotation). At each angle the user clicks the sample centre in the image and presses **Submit**. After all 8 angles the data is sent to the **Fit** tab of the GUI where a sinusoidal fit runs automatically in both x and y directions. The fit parameters are loaded automatically at the end of the procedure.
@@ -84,12 +82,10 @@ The sample fine alignment can be obtained using ptychography. For this a short l
`lamni.read_additional_correction('/sls/X12SA/data/e20632/Data10/cxs_software/ptycho/correction_lamni_um_S05389_lamni_fit.txt')`
* With this alignment a second iteration could be performed. To read the second correction file use `lamni.read_additional_correction_2()`
#### Shifting the FOV
* `lamni.tomo_fovx/y_offset=value` [mm] will shift the field of view. Perform this adjustment from projections collected at **lsamrot 0 degrees**. This shift will rotate. In contrast the manual shift will be a constant shift, identical at all angles.
(user.ptychography.lamni.laminography)=
### Laminography scan
### Test projections and Laminography scan
Start the laminography scan by
@@ -101,7 +97,7 @@ Start the laminography scan by
During the scan, a live progress report is printed (subtomogram/projection counters and an estimated time of completion). Gaps significantly longer than a normal acquisition cycle (e.g. a beamline-down interruption, or a crash followed by a restart) are detected automatically from the time between consecutive projections and excluded from the time estimate; the total time lost to such gaps is printed once the scan finishes.
Three angular sampling modes are implemented, same as flOMNI, but LamNI always scans the full 360 degrees — laminography's tilted sample geometry does not have the symmetry that lets flOMNI's 180-degree mode work, so there is no angular-range choice to make:
Three angular sampling modes are implemented:
| tomography mode | parameters and defaults |
| --- | --- |
@@ -111,13 +107,13 @@ Three angular sampling modes are implemented, same as flOMNI, but LamNI always s
The parameters above can be used to __restart an interrupted acquisition__ manually, or - more conveniently - by running
`lamni.tomo_scan_resume()`
which reads the last recorded progress and resumes automatically at the exact point (subtomogram/angle, or projection for the golden ratio modes) the scan was interrupted at, without needing to look up the values by hand. When running from the tomo scan queuing system use `lamni.tomo_queue_execute()` instead!
which reads the last recorded progress and resumes automatically at the exact point (subtomogram/angle, or projection for the golden ratio modes) the scan was interrupted at, without needing to look up the values by hand. When running from the tomo scan queuing system use `lamni.tomo_queue_resume()` instead!
In case of eight equally spaced sub-tomograms, an individual sub-tomogram can be scanned by `lamni.sub_tomo_scan(subtomo_number, start_angle)`. If the start angle is not specified, it is computed depending on `subtomo_number`, ranging from 1 to 8.
#### Queueing multiple scans
Several tomo parameter sets can be queued and run sequentially on the same sample, without having to start each one by hand.
Several tomo parameter sets can be queued and run sequentially on the same sample, without having to start each one by hand. Easiest is to use the tomo parameters GUI for preparing the scans. But it is also possible on the command line interface.
| command | explanation |
| --- | --- |
@@ -149,20 +145,6 @@ as if you'd just clicked Edit. Nothing is written yet — review or tweak the fi
then Submit (writes live, blocked while the beamline is busy) or "Add to queue" (always
allowed) as usual. If you already had an edit in progress, it asks before discarding it.
**GUI: two different "Add to queue" buttons.** The params panel's own **"Add to
queue"** (visible in edit mode) queues whatever you've typed, unsubmitted. The ☰
Queue control… window's **"Add current params to queue"** is a different button on a
different window — it always queues the *live* parameters, regardless of any edit
open in the panel. If you have an unsaved edit open and click the queue window's
button instead of the panel's, it warns and names the correct one before proceeding,
since it would otherwise queue your last-submitted values, not what you just typed.
**GUI: "Duplicate selected".** Appends an exact copy of the selected job (any status —
pending, running, incomplete, or done, and either a tomo or a command job) to the end
of the queue, labeled with `_dup` appended, status reset to `pending`. Useful for
re-running a job with the same settings without re-typing them, or as a starting point
to tweak via "Load into editor".
**Reacquiring from an earlier projection.** A resumed job normally picks up
exactly where it stopped. If a beamline problem means earlier projections need
redoing too — even in a job that's already marked `done` — use:
@@ -1,7 +1,10 @@
import builtins
import types
import numpy as np
import pytest
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni, _ProgressProxy
plan = Flomni._subtomo_angle_plan
@@ -134,3 +137,121 @@ def _actual_grid(stepsize):
N = int(180.0 / stepsize)
step = 180.0 / N
return N, step, N * 8
# ── golden-ratio (tomo_type 2/3) resume must not fire a spurious 0-deg shot ──
#
# Same bug/fix as LamNI.tomo_scan() (see test_lamni_tomo_angles.py):
# previous_subtomo_number starts at -1 on every tomo_scan() call, including a
# resume (projection_number set). If the resume lands on an odd sub-tomogram
# past projection 10, the very first loop iteration looks like a
# "transition" and used to fire the 0-deg damage-estimation shot even though
# the rotation isn't actually passing through 0 deg at that moment.
class _FakeClient:
def __init__(self):
self._vars = {}
def get_global_var(self, key):
return self._vars.get(key)
def set_global_var(self, key, value):
self._vars[key] = value
class _FakeContextManager:
def __enter__(self):
return self
def __exit__(self, *exc_info):
return False
class _FakeScans:
dataset_id_on_hold = _FakeContextManager()
def _fake_golden(ii, bunch_size, *args, **kwargs):
if ii < 11:
return float(ii), 1
if ii < 20:
return float(ii), 3
return float(ii), 5
def make_flomni_for_golden_resume(monkeypatch, tomo_type: int) -> Flomni:
"""Bare Flomni instance with tomo_scan()'s golden-ratio (tomo_type 2/3)
branch reachable, everything else (hardware checks, GUI, sample DB,
scilog) stubbed out so only the resume-guard logic under test runs."""
obj = object.__new__(Flomni)
obj.client = _FakeClient()
obj._progress_proxy = _ProgressProxy(obj.client)
obj.tomo_type = tomo_type
obj.tomo_angle_stepsize = 45.0
obj.golden_ratio_bunch_size = 20
obj.golden_projections_at_0_deg_for_damage_estimation = 1
obj.golden_max_number_of_projections = 21
obj.tomo_id = -1
obj.sample_get_name = lambda position: "test" # backs the read-only sample_name property
obj.special_angles = []
obj._golden = _fake_golden
obj._golden_equally_spaced = _fake_golden
obj._check_eye_out_and_optics_in = lambda: True
obj.flomnigui_show_progress = lambda: None
obj.add_sample_database = lambda *a, **k: 0
obj.write_pdf_report = lambda: None
obj.collect_empty_frames = lambda: None
obj._print_progress = lambda: None
obj._format_duration = lambda seconds: "0s"
obj._describe_active_hook = lambda: None
obj._active_hook_source = lambda: None
obj.at_each_angle_hook = None
obj._scilog_write = lambda content, tag: None
obj.OMNYTools = types.SimpleNamespace(printgreenbold=lambda msg: None)
fake_bec = types.SimpleNamespace(
active_account="",
queue=types.SimpleNamespace(next_scan_number=1),
builtin_actors=types.SimpleNamespace(
scan_interlock=types.SimpleNamespace(trigger_setting=None, enabled=False)
),
)
monkeypatch.setitem(builtins.__dict__, "bec", fake_bec)
monkeypatch.setitem(builtins.__dict__, "scans", _FakeScans())
return obj
@pytest.mark.parametrize("tomo_type", [2, 3])
def test_golden_resume_suppresses_spurious_zero_deg_shot(monkeypatch, tomo_type):
flomni = make_flomni_for_golden_resume(monkeypatch, tomo_type)
recorded = []
flomni._tomo_scan_at_angle = lambda angle, subtomo: recorded.append((angle, subtomo))
flomni.tomo_scan(projection_number=11) # resumes exactly at the subtomo-3 transition
assert (0, 3) not in recorded, "resume must not fire the 0-deg shot at its own start"
@pytest.mark.parametrize("tomo_type", [2, 3])
def test_golden_resume_still_fires_zero_deg_shot_at_later_genuine_transition(
monkeypatch, tomo_type
):
flomni = make_flomni_for_golden_resume(monkeypatch, tomo_type)
recorded = []
flomni._tomo_scan_at_angle = lambda angle, subtomo: recorded.append((angle, subtomo))
flomni.tomo_scan(projection_number=11)
assert (0, 5) in recorded, "a later genuine transition within the same resumed call must still fire"
@pytest.mark.parametrize("tomo_type", [2, 3])
def test_golden_fresh_start_still_fires_zero_deg_shot_at_natural_transition(monkeypatch, tomo_type):
flomni = make_flomni_for_golden_resume(monkeypatch, tomo_type)
recorded = []
flomni._tomo_scan_at_angle = lambda angle, subtomo: recorded.append((angle, subtomo))
flomni.tomo_scan() # projection_number=None -- a fresh, non-resumed run
assert (0, 3) in recorded, "a fresh (non-resume) run must still fire the shot at the natural transition"
@@ -44,6 +44,7 @@ class FakeDev:
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.leyey = FakeAxis(5.0, in_pos=0.0, out_pos=5.0)
self.rtx = types.SimpleNamespace(enabled=rtx_enabled, controller=FakeController())
def __contains__(self, name):
@@ -141,3 +142,30 @@ def test_losa_out_moves_when_not_out(fake_dev, umv_calls):
(fake_dev.losaz, -1.0),
(fake_dev.losay, 1.0),
]
def test_check_eye_out_and_optics_in_true_when_ready(fake_dev):
optics = make_optics()
assert optics._check_eye_out_and_optics_in() == True
def test_check_eye_out_and_optics_in_false_when_eye_not_out(fake_dev):
fake_dev.leyey.value = 0.0 # at "in" (0.0), not "out" (5.0)
optics = make_optics()
assert optics._check_eye_out_and_optics_in() == False
def test_check_eye_out_and_optics_in_false_when_fzp_not_in(fake_dev):
fake_dev.loptx.value = -1.0 # at "out", not "in" (-0.5)
optics = make_optics()
assert optics._check_eye_out_and_optics_in() == False
def test_check_eye_out_and_optics_in_false_when_osa_not_in(fake_dev):
fake_dev.losaz.value = -1.0 # at "out", not "in" (1.0)
optics = make_optics()
assert optics._check_eye_out_and_optics_in() == False
@@ -170,9 +170,10 @@ def make_lamni_for_tomo_scan(
obj.at_each_angle_hook = None
obj.OMNYTools = types.SimpleNamespace(printgreenbold=lambda msg: None)
# These tests exercise tomo_scan()'s account-handling/heartbeat/progress-GUI
# logic, not the fine-alignment confirmation gate -- bypass it so it never
# blocks on input().
# logic, not the fine-alignment confirmation gate or the measurement-
# configuration check -- bypass both so they never block on input().
obj._confirm_sequence_override = lambda *a, **k: True
obj._check_eye_out_and_optics_in = lambda: True
monkeypatch.setitem(
builtins.__dict__,
"bec",
@@ -348,3 +349,115 @@ def test_zero_deg_reference_skipped_when_resuming_mid_subtomo(monkeypatch):
lamni.tomo_scan(subtomo_start=3, start_angle=45.0)
assert recorded == [(0, 5), (0, 7), (0, 8)]
# ── tomo_scan()'s measurement-configuration check (eye out, FZP/OSA in) ─────
#
# Ported from Flomni.tomo_scan()'s equivalent guard
# (_check_eye_out_and_optics_in()) -- LamNI had no such check at all.
def test_tomo_scan_aborts_when_not_ready_and_declined(monkeypatch):
lamni = make_lamni_for_tomo_scan(monkeypatch, 45.0, active_account="")
lamni._check_eye_out_and_optics_in = lambda: False
lamni._confirm_sequence_override = lambda *a, **k: False
calls = []
lamni.add_sample_database = lambda *a, **k: calls.append("registered") or 0
lamni.tomo_scan()
assert calls == []
def test_tomo_scan_proceeds_when_not_ready_but_confirmed(monkeypatch):
lamni = make_lamni_for_tomo_scan(monkeypatch, 45.0, active_account="")
lamni._check_eye_out_and_optics_in = lambda: False
lamni._confirm_sequence_override = lambda *a, **k: True
lamni.add_sample_database = lambda *a, **k: 0
lamni.tomo_scan() # must not raise / must reach the rest of the scan
def test_tomo_scan_non_interactive_warns_and_proceeds_without_input(monkeypatch):
"""tomo_queue_execute()'s unattended runs pass interactive=False -- the
check must never call input(), only warn and proceed after a delay."""
lamni = make_lamni_for_tomo_scan(monkeypatch, 45.0, active_account="")
lamni._check_eye_out_and_optics_in = lambda: False
lamni._confirm_sequence_override = lambda *a, **k: (_ for _ in ()).throw(
AssertionError("must not prompt when interactive=False")
)
warnings = []
lamni.OMNYTools = types.SimpleNamespace(
printredbold=lambda msg: warnings.append(msg), printgreenbold=lambda msg: None
)
lamni.add_sample_database = lambda *a, **k: 0
monkeypatch.setattr("csaxs_bec.bec_ipython_client.plugins.LamNI.lamni.time.sleep", lambda s: None)
lamni.tomo_scan(interactive=False)
assert any("measurement configuration" in w for w in warnings)
# ── golden-ratio (tomo_type 2/3) resume must not fire a spurious 0-deg shot ──
#
# previous_subtomo_number starts at -1 on every tomo_scan() call, including a
# resume (projection_number set). If the resume lands on an odd sub-tomogram
# past projection 10, the very first loop iteration looks like a "transition"
# and used to fire the 0-deg damage-estimation shot even though the rotation
# isn't actually passing through 0 deg at that moment -- mirrors tomo_type 1's
# already-fixed start_angle-is-None resume guard.
def _fake_golden(ii, bunch_size, maxangle=360, reverse=True):
if ii < 11:
return float(ii), 1
if ii < 20:
return float(ii), 3
return float(ii), 5
def make_lamni_for_golden_resume(monkeypatch, tomo_type: int) -> LamNI:
lamni = make_lamni_for_tomo_scan(monkeypatch, 45.0, active_account="")
lamni.tomo_type = tomo_type
lamni.golden_ratio_bunch_size = 20
lamni.golden_projections_at_0_deg_for_damage_estimation = 1
lamni.golden_max_number_of_projections = 21
lamni.add_sample_database = lambda *a, **k: 0
lamni._golden = _fake_golden
lamni._golden_equally_spaced = _fake_golden
return lamni
@pytest.mark.parametrize("tomo_type", [2, 3])
def test_golden_resume_suppresses_spurious_zero_deg_shot(monkeypatch, tomo_type):
lamni = make_lamni_for_golden_resume(monkeypatch, tomo_type)
recorded = []
lamni._tomo_scan_at_angle = lambda angle, subtomo: recorded.append((angle, subtomo))
lamni.tomo_scan(projection_number=11) # resumes exactly at the subtomo-3 transition
assert (0, 3) not in recorded, "resume must not fire the 0-deg shot at its own start"
@pytest.mark.parametrize("tomo_type", [2, 3])
def test_golden_resume_still_fires_zero_deg_shot_at_later_genuine_transition(
monkeypatch, tomo_type
):
lamni = make_lamni_for_golden_resume(monkeypatch, tomo_type)
recorded = []
lamni._tomo_scan_at_angle = lambda angle, subtomo: recorded.append((angle, subtomo))
lamni.tomo_scan(projection_number=11)
assert (0, 5) in recorded, "a later genuine transition within the same resumed call must still fire"
@pytest.mark.parametrize("tomo_type", [2, 3])
def test_golden_fresh_start_still_fires_zero_deg_shot_at_natural_transition(monkeypatch, tomo_type):
lamni = make_lamni_for_golden_resume(monkeypatch, tomo_type)
recorded = []
lamni._tomo_scan_at_angle = lambda angle, subtomo: recorded.append((angle, subtomo))
lamni.tomo_scan() # projection_number=None -- a fresh, non-resumed run
assert (0, 3) in recorded, "a fresh (non-resume) run must still fire the shot at the natural transition"
@@ -0,0 +1,47 @@
"""Regression test for LamniGalilController.lgalil_is_air_off_and_orchestra_enabled().
The method used to do `bool(self.socket_put_and_receive(...))` directly on the
raw Galil "MG" string reply (e.g. "0.0000"), which is always truthy in Python
regardless of the actual register value -- the method could never report
anything but True. Every other digital-register read in this module goes
through float(...) first; this was the one place missing it.
"""
import pytest
from csaxs_bec.devices.omny.galil.lgalil_ophyd import LamniGalilController
from csaxs_bec.devices.sim.sim_lamni import SimLamniGalilMotor
from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
HOST = "mpc2680.psi.ch"
PORT = 8081
@pytest.fixture(autouse=True)
def reset_sim_state():
"""Avoid cross-test leakage of shared (host, port)-keyed simulation state."""
SimStateRegistry.reset()
yield
SimStateRegistry.reset()
@pytest.fixture
def lsamrot(dm_with_devices):
LamniGalilController._reset_controller()
motor = SimLamniGalilMotor("C", name="lsamrot", host=HOST, port=PORT, device_manager=dm_with_devices)
motor.controller.on()
yield motor
motor.controller.off()
motor.controller._reset_controller()
def test_lgalil_is_air_off_and_orchestra_enabled_reflects_registers(lsamrot):
# sim default: settled (air off, RT feedback not blocked)
assert lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled() is True
lsamrot.controller.socket_put_confirmed("CB9") # RT feedback blocked
assert lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled() is False
lsamrot.controller.socket_put_confirmed("SB9")
lsamrot.controller.socket_put_confirmed("CB13") # air back on
assert lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled() is False
@@ -0,0 +1,45 @@
from unittest import mock
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
def _scan_stub(user_setpoint: float, readback: float, tolerance: float = 0.5):
"""A minimal stand-in for a FlomniFermatScan instance, isolated to just the
attributes flomni_rotation() touches (self.dev.fsamroy, self.actions).
"""
scan = mock.MagicMock()
scan.scan_info.metadata = {"RID": "rid-test"}
scan.dev.fsamroy.user_setpoint.get.return_value = user_setpoint
scan.dev.fsamroy.readback.get.return_value = readback
scan.dev.fsamroy.tolerance = tolerance
return scan
def test_flomni_rotation_moves_when_setpoint_differs():
scan = _scan_stub(user_setpoint=5.0, readback=10.0, tolerance=0.5)
FlomniFermatScan.flomni_rotation(scan, 10.0)
scan.actions.set.assert_called_once()
args, kwargs = scan.actions.set.call_args
assert args[0] is scan.dev.fsamroy
assert args[1] == 10.0
assert kwargs.get("wait") is False
def test_flomni_rotation_skipped_when_settled_at_target():
scan = _scan_stub(user_setpoint=10.0, readback=10.1, tolerance=0.5)
FlomniFermatScan.flomni_rotation(scan, 10.0)
scan.actions.set.assert_not_called()
def test_flomni_rotation_moves_when_setpoint_matches_but_readback_far_off():
# simulates a previous rotation interrupted before the axis actually
# reached the target: the setpoint cache matches, the readback doesn't
scan = _scan_stub(user_setpoint=10.0, readback=2.0, tolerance=0.5)
FlomniFermatScan.flomni_rotation(scan, 10.0)
scan.actions.set.assert_called_once()
@@ -0,0 +1,46 @@
from unittest import mock
from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan
def _scan_stub(user_setpoint: float, orchestra_enabled: bool):
"""A minimal stand-in for a LamniFermatScan instance, isolated to just the
attributes lamni_rotation() touches (self.dev.lsamrot, self.actions).
"""
scan = mock.MagicMock()
scan.dev.lsamrot.user_setpoint.get.return_value = user_setpoint
scan.dev.lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled.return_value = (
orchestra_enabled
)
return scan
def test_lamni_rotation_moves_when_setpoint_differs():
scan = _scan_stub(user_setpoint=5.0, orchestra_enabled=True)
LamniFermatScan.lamni_rotation(scan, 10.0)
# a genuine target change must move regardless of the orchestra-enabled
# state, and must not even need to query it (short-circuit)
scan.dev.lsamrot.controller.lgalil_is_air_off_and_orchestra_enabled.assert_not_called()
scan.dev.lsamrot.set.assert_called_once_with(10.0)
scan.dev.lsamrot.set.return_value.wait.assert_called_once_with()
def test_lamni_rotation_skipped_when_settled_at_target():
scan = _scan_stub(user_setpoint=10.0, orchestra_enabled=True)
LamniFermatScan.lamni_rotation(scan, 10.0)
scan.dev.lsamrot.set.assert_not_called()
def test_lamni_rotation_moves_when_setpoint_matches_but_not_settled():
# simulates a previous rotation interrupted before the air bearings
# clamped and the piezo stage was released back to interferometer feedback
scan = _scan_stub(user_setpoint=10.0, orchestra_enabled=False)
LamniFermatScan.lamni_rotation(scan, 10.0)
scan.dev.lsamrot.set.assert_called_once_with(10.0)
scan.dev.lsamrot.set.return_value.wait.assert_called_once_with()
@@ -0,0 +1,44 @@
from unittest import mock
from csaxs_bec.scans.omny_fermat_scan import OmnyFermatScan
def _scan_stub(user_setpoint: float, readback: float, tolerance: float = 0.5):
"""A minimal stand-in for an OmnyFermatScan instance, isolated to just the
attributes omny_rotation() touches (self.dev.osamroy, self.actions).
"""
scan = mock.MagicMock()
scan.dev.osamroy.user_setpoint.get.return_value = user_setpoint
scan.dev.osamroy.readback.get.return_value = readback
scan.dev.osamroy.tolerance = tolerance
return scan
def test_omny_rotation_moves_when_setpoint_differs():
scan = _scan_stub(user_setpoint=5.0, readback=10.0, tolerance=0.5)
OmnyFermatScan.omny_rotation(scan, 10.0)
scan.actions.set.assert_called_once()
args, kwargs = scan.actions.set.call_args
assert args[0] is scan.dev.osamroy.user_setpoint
assert args[1] == 10.0
assert kwargs.get("wait") is False
def test_omny_rotation_skipped_when_settled_at_target():
scan = _scan_stub(user_setpoint=10.0, readback=10.1, tolerance=0.5)
OmnyFermatScan.omny_rotation(scan, 10.0)
scan.actions.set.assert_not_called()
def test_omny_rotation_moves_when_setpoint_matches_but_readback_far_off():
# simulates a previous rotation interrupted before the axis actually
# reached the target: the setpoint cache matches, the readback doesn't
scan = _scan_stub(user_setpoint=10.0, readback=2.0, tolerance=0.5)
OmnyFermatScan.omny_rotation(scan, 10.0)
scan.actions.set.assert_called_once()