feat/warn early about Fermat scans below the minimum point count

FlomniFermatScan/LamNIFermatScan already refuse to run with fewer than
20 positions (_check_min_positions(), raises ScanAbortion), but only
once the scan actually starts -- too late for an unattended tomo-queue
run, where the job just aborts mid-run and pauses the queue. Surface
the same prediction earlier instead, as a non-blocking warning:

- Extracted the position-generating methods on both scan classes
  (get_flomni_fermat_spiral_pos, and lamni's chain --
  _lamni_compute_scan_center, _lamni_compute_stitch_center,
  _compute_total_shift, _lamni_check_pos_in_fov_range_and_circ_fov,
  get_lamni_fermat_spiral_pos) into pure @staticmethods, so the exact
  same algorithm the scan server runs can also be called from
  client-side code without a live scan session. Behavior-preserving --
  verified against the existing exact-position/instruction assertion
  tests for both classes. Lifted the hardcoded "20" into a _MIN_POSITIONS
  class attribute on each, so client code references the same threshold.
- lamni.py/flomni.py: new _expected_fermat_position_count() calls the
  real scan-class algorithm with the live tomo parameters and prints a
  warning line in tomo_parameters() when below the minimum.
- tomo_params.py: a new live-updating "Estimated Fermat scan points"
  field (mirroring the existing achievable-step preview's styling),
  wired to the fov/step/stitch/piezo-range fields, flagged orange below
  the minimum -- never blocks Submit/Add-to-queue.

lamni's circular-FOV crop is angle/stitch-dependent, so the estimate is
representative (center tile, angle 0 for the CLI; the live stitch tile
for the GUI, which has those fields right there) rather than an exact
per-projection guarantee -- documented as such in both places.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
x01dc
2026-07-21 03:52:44 +02:00
co-authored by Claude Sonnet 5
parent 4ada8e85ce
commit 8a7a350280
8 changed files with 581 additions and 43 deletions
@@ -1288,6 +1288,51 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
# Parameter display and interactive update
# ------------------------------------------------------------------
def _expected_fermat_position_count(self) -> int:
"""Predict the number of Fermat-spiral scan positions the current
settings would produce per projection tile, using the exact same
algorithm LamNIFermatScan runs at scan time
(LamNIFermatScan.get_lamni_fermat_spiral_pos()) -- so a
too-few-points configuration (which the scan server would only
catch by aborting with ScanAbortion once the scan actually starts)
can be caught here instead, while just looking at tomo_parameters().
center_x/center_y/shift_x/shift_y are deliberately left at their
defaults (0): they only shift the final absolute position, applied
after the FOV/circular-FOV keep-or-discard check, so they can never
affect the resulting count. Computed for the center stitch tile
(stitch_x=stitch_y=0) at angle=0, as a representative estimate --
the real per-projection count can vary a bit with angle/stitch
tile, since the circular FOV crop (tomo_circfov) is checked against
the rotated stage position.
"""
from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan
positions = LamNIFermatScan.get_lamni_fermat_spiral_pos(
-abs(self.lamni_piezo_range_x / 2),
abs(self.lamni_piezo_range_x / 2),
-abs(self.lamni_piezo_range_y / 2),
abs(self.lamni_piezo_range_y / 2),
step=self.tomo_shellstep,
spiral_type=0,
center=False,
angle=0.0,
stitch_x=0,
stitch_y=0,
stitch_overlap=self.tomo_stitch_overlap,
fov_size=[self.lamni_piezo_range_x, self.lamni_piezo_range_y],
fov_circular=self.tomo_circfov,
)
return len(positions)
@staticmethod
def _fermat_min_positions() -> int:
"""LamNIFermatScan's own minimum-position threshold -- see
_expected_fermat_position_count()."""
from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan
return LamNIFermatScan._MIN_POSITIONS
def tomo_parameters(self):
"""Print and interactively update the tomo parameters."""
print("Current settings:")
@@ -1300,6 +1345,15 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
print(f"Stitching number x,y = {self.lamni_stitch_x}, {self.lamni_stitch_y}")
print(f"Stitching overlap = {self.tomo_stitch_overlap}")
print(f"Circular FOV diam <microns> = {self.tomo_circfov}")
expected_positions = self._expected_fermat_position_count()
min_positions = self._fermat_min_positions()
if expected_positions < min_positions:
print(
f"Estimated Fermat scan points per projection: {expected_positions} "
f"(WARNING: below the minimum of {min_positions} -- the scan will abort when run)"
)
else:
print(f"Estimated Fermat scan points per projection: {expected_positions}")
print(f"Reconstruction queue name = {self.ptycho_reconstruct_foldername}")
print(f"Frames per trigger (burst) = {self.frames_per_trigger}")
print("FOV offset rotates to find the ROI; initial values determined in Xrayeye alignment.")
@@ -3393,6 +3393,40 @@ class Flomni(
step = 180.0 / N
return N, step, N * 8
def _expected_fermat_position_count(self) -> int:
"""Predict the number of Fermat-spiral scan positions the current
settings would produce per projection tile, using the exact same
algorithm FlomniFermatScan runs at scan time
(FlomniFermatScan.get_flomni_fermat_spiral_pos()) -- so a
too-few-points configuration (which the scan server would only
catch by aborting with ScanAbortion once the scan actually starts)
can be caught here instead, while just looking at tomo_parameters().
cenx/ceny/zshift are deliberately left at their defaults: they only
shift the final position, applied after the FOV keep-or-discard
check, so they can never affect the resulting count.
"""
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
positions = FlomniFermatScan.get_flomni_fermat_spiral_pos(
-abs(self.fovx / 2),
abs(self.fovx / 2),
-abs(self.fovy / 2),
abs(self.fovy / 2),
step=self.tomo_shellstep,
spiral_type=0,
center=False,
)
return len(positions)
@staticmethod
def _fermat_min_positions() -> int:
"""FlomniFermatScan's own minimum-position threshold -- see
_expected_fermat_position_count()."""
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
return FlomniFermatScan._MIN_POSITIONS
def tomo_parameters(self):
"""print and update the tomo parameters"""
print("Current settings:")
@@ -3401,6 +3435,17 @@ class Flomni(
print(f"FOV (200/100) <microns> = {self.fovx}, {self.fovy}")
print(f"Stitching number x,y = {self.stitch_x}, {self.stitch_y}")
print(f"Stitching overlap = {self.tomo_stitch_overlap}")
if not self.single_point_instead_of_fermat_scan:
expected_positions = self._expected_fermat_position_count()
min_positions = self._fermat_min_positions()
if expected_positions < min_positions:
print(
f"Estimated Fermat scan points per projection: {expected_positions} "
f"(WARNING: below the minimum of {min_positions} -- the scan will abort"
" when run)"
)
else:
print(f"Estimated Fermat scan points per projection: {expected_positions}")
print(f"Reconstruction queue name = {self.ptycho_reconstruct_foldername}")
print(f" _manual_shift_y <um> = {self.manual_shift_y}")
print(f"Frames per trigger (burst) = {self.frames_per_trigger}")
@@ -481,6 +481,17 @@ class TomoParamsWidget(BECWidget, QWidget):
}
for token in self._profile["field_order"]:
token_builders[token]()
# Estimated Fermat-scan point count -- live preview, updates as the
# fov/step/stitch/piezo-range fields above change, flagged orange if
# below the scan server's own minimum (see _update_fermat_position_preview()).
self._lbl_fermat_positions = QLabel("---")
common_form.addRow("Estimated Fermat scan points:", self._lbl_fermat_positions)
for key in self._profile["fermat_position_fields"]:
widget = self._pw.get(key)
if widget is not None:
widget.valueChanged.connect(self._update_fermat_position_preview)
vbox.addLayout(common_form)
# type-1 section
@@ -732,6 +743,7 @@ class TomoParamsWidget(BECWidget, QWidget):
self._pw["_requested_total"].blockSignals(False)
self._update_projection_preview()
self._update_fermat_position_preview()
# type-3: derive projections-per-subtomo from stored tomo_angle_stepsize
# (base angle is 180 for flomni, 360 for lamni -- see
@@ -888,6 +900,30 @@ class TomoParamsWidget(BECWidget, QWidget):
self._lbl_actual_total.setStyleSheet("")
self._lbl_actual_total.setToolTip("")
def _update_fermat_position_preview(self) -> None:
"""Live estimate of the Fermat-scan point count for the currently
edited fov/step/stitch/piezo-range fields (see
self._profile["compute_fermat_positions"], which calls the real
scan class's own position-generation algorithm -- not a
reimplementation). Warning-only: flags orange below the scan
server's own minimum, never blocks Submit/Add-to-queue."""
params = {}
for key in self._profile["fermat_position_fields"]:
widget = self._pw.get(key)
if widget is not None:
params[key] = widget.value()
count, min_positions = self._profile["compute_fermat_positions"](params)
self._lbl_fermat_positions.setText(str(count))
if count < min_positions:
self._lbl_fermat_positions.setStyleSheet("color: orange;")
self._lbl_fermat_positions.setToolTip(
f"Below the minimum of {min_positions} -- the scan will abort when run."
)
else:
self._lbl_fermat_positions.setStyleSheet("")
self._lbl_fermat_positions.setToolTip("")
# ── slots ─────────────────────────────────────────────────────────────────
def _on_type_changed(self, _index: int) -> None:
@@ -2089,6 +2125,72 @@ def _lamni_compute_type1(angle_range: int, stepsize: float) -> tuple[int, float,
return actual_total, achievable_step, stepsize
def _compute_fermat_positions_flomni(params: dict[str, Any]) -> tuple[int, int]:
"""Estimated Fermat-spiral point count for the currently edited flomni
fields, plus the scan-server's own minimum threshold.
Unlike _compute_type1() above, this does NOT reimplement the spiral
math -- it calls FlomniFermatScan.get_flomni_fermat_spiral_pos()
directly (a pure @staticmethod, no device I/O), the exact same
algorithm the scan itself runs, so the estimate can never drift out of
sync with what actually happens at scan time. cenx/ceny/zshift are left
at their defaults: they only shift the final position, applied after
the FOV keep-or-discard check, so they never affect the count. Stitch
doesn't affect flomni's count either -- every stitched tile is an
identically-shaped Fermat scan, just centered differently.
Returns:
(estimated_count, minimum_required)
"""
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
fovx = params.get("fovx", 0.0)
fovy = params.get("fovy", 0.0)
step = params.get("tomo_shellstep", 0.0)
if step <= 0:
return 0, FlomniFermatScan._MIN_POSITIONS
positions = FlomniFermatScan.get_flomni_fermat_spiral_pos(
-abs(fovx / 2), abs(fovx / 2), -abs(fovy / 2), abs(fovy / 2), step=step, spiral_type=0
)
return len(positions), FlomniFermatScan._MIN_POSITIONS
def _compute_fermat_positions_lamni(params: dict[str, Any]) -> tuple[int, int]:
"""Lamni sibling of _compute_fermat_positions_flomni(): calls
LamNIFermatScan.get_lamni_fermat_spiral_pos() directly -- including its
rotated-stage and circular-FOV (tomo_circfov) cropping, which can matter
a lot and would be easy to get subtly wrong in a reimplementation.
Evaluated at the *currently edited* stitch tile and angle=0 (lamni has
no angle field in this widget -- the real per-projection count can vary
a bit with angle, since the circular crop is angle-dependent).
Returns:
(estimated_count, minimum_required)
"""
from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan
piezo_x = params.get("lamni_piezo_range_x", 0.0)
piezo_y = params.get("lamni_piezo_range_y", 0.0)
step = params.get("tomo_shellstep", 0.0)
if step <= 0:
return 0, LamNIFermatScan._MIN_POSITIONS
positions = LamNIFermatScan.get_lamni_fermat_spiral_pos(
-abs(piezo_x / 2),
abs(piezo_x / 2),
-abs(piezo_y / 2),
abs(piezo_y / 2),
step=step,
spiral_type=0,
angle=0.0,
stitch_x=params.get("lamni_stitch_x", 0),
stitch_y=params.get("lamni_stitch_y", 0),
stitch_overlap=params.get("tomo_stitch_overlap", 1.0),
fov_size=[piezo_x, piezo_y],
fov_circular=params.get("tomo_circfov", 0.0),
)
return len(positions), LamNIFermatScan._MIN_POSITIONS
def _lamni_get_tomo_fov_offset(client, axis: str) -> float:
"""Mirrors LamNIAlignmentMixin.tomo_fovx_offset/tomo_fovy_offset getters
exactly: both axes are packed into one global var, ``tomo_fov_offset =
@@ -2155,6 +2257,8 @@ SETUP_PROFILES: dict[str, dict[str, Any]] = {
"type1_base_angle": 180.0,
"compute_type1": _compute_type1,
"requested_to_stepsize": _requested_to_stepsize,
"compute_fermat_positions": _compute_fermat_positions_flomni,
"fermat_position_fields": ["tomo_shellstep", "fovx", "fovy"],
"sample_name_getter": lambda w: w.dev.flomni_samples.sample_names.sample0.get(),
"cli_hint_name": "flomni",
},
@@ -2205,6 +2309,16 @@ SETUP_PROFILES: dict[str, dict[str, Any]] = {
"type1_base_angle": 360.0,
"compute_type1": _lamni_compute_type1,
"requested_to_stepsize": _lamni_requested_to_stepsize,
"compute_fermat_positions": _compute_fermat_positions_lamni,
"fermat_position_fields": [
"tomo_shellstep",
"tomo_circfov",
"lamni_stitch_x",
"lamni_stitch_y",
"tomo_stitch_overlap",
"lamni_piezo_range_x",
"lamni_piezo_range_y",
],
"sample_name_getter": lambda w: w.client.get_global_var("sample_name"),
"cli_hint_name": "lamni",
},
+92 -28
View File
@@ -49,7 +49,8 @@ def lamni_from_stage_coordinates(x_stage: float, y_stage: float) -> tuple:
class LamNIMixin:
def _lamni_compute_scan_center(self, x, y, angle_deg):
@staticmethod
def _lamni_compute_scan_center(x, y, angle_deg):
# assuming a scan point was found at interferometer x,y at zero degrees
# this function computes the new interferometer coordinates of this spot
# at a different rotation angle based on the lamni geometry
@@ -216,6 +217,13 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
arg_input = {}
arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None}
# Minimum number of Fermat-spiral positions a scan is allowed to run
# with -- exposed as a class attribute (not just a literal inside
# _check_min_positions()) so client-side code (tomo_parameters(),
# tomo_params.py) can warn about a too-few-points configuration before
# it ever reaches the scan server, using the exact same threshold.
_MIN_POSITIONS = 20
def __init__(self, *args, parameter: dict = None, frames_per_trigger:int=1, exp_time:float=0,**kwargs):
"""
A LamNI scan following Fermat's spiral.
@@ -286,20 +294,24 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
self._check_min_positions()
def _check_min_positions(self):
if self.num_pos < 20:
if self.num_pos < self._MIN_POSITIONS:
raise ScanAbortion(
f"The number of positions must exceed 20. Currently: {self.num_pos}."
f"The number of positions must exceed {self._MIN_POSITIONS}. Currently:"
f" {self.num_pos}."
)
def _lamni_check_pos_in_fov_range_and_circ_fov(self, x, y) -> bool:
@staticmethod
def _lamni_check_pos_in_fov_range_and_circ_fov(
x, y, stitch_x, stitch_y, angle, fov_size, stitch_overlap, fov_circular
) -> bool:
# this function checks if positions are reachable in a scan
# these x y intererometer positions are not shifted to the scan center
# so its purpose is to see if the position is reachable by the
# rotated piezo stage. For a scan these positions have to be shifted to
# the current scan center before starting the scan
stage_x, stage_y = lamni_to_stage_coordinates(x, y)
stage_x_with_stitch, stage_y_with_stitch = self._lamni_compute_stitch_center(
self.stitch_x, self.stitch_y, self.angle
stage_x_with_stitch, stage_y_with_stitch = LamNIFermatScan._lamni_compute_stitch_center(
stitch_x, stitch_y, angle, fov_size, stitch_overlap
)
stage_x_with_stitch, stage_y_with_stitch = lamni_to_stage_coordinates(
stage_x_with_stitch, stage_y_with_stitch
@@ -307,7 +319,7 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
# piezo stage is currently rotated to stage_angle_deg in degrees
# rotate positions to the piezo stage system
alpha = (self.angle - 300 + 30.5) / 180 * np.pi
alpha = (angle - 300 + 30.5) / 180 * np.pi
stage_x_rot = np.cos(alpha) * stage_x + np.sin(alpha) * stage_y
stage_y_rot = -np.sin(alpha) * stage_x + np.cos(alpha) * stage_y
@@ -319,22 +331,32 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
)
return (
np.abs(stage_x_rot) <= (self.fov_size[1] / 2)
and np.abs(stage_y_rot) <= (self.fov_size[0] / 2)
np.abs(stage_x_rot) <= (fov_size[1] / 2)
and np.abs(stage_y_rot) <= (fov_size[0] / 2)
and (
self.fov_circular == 0
fov_circular == 0
or (
np.power((stage_x_rot_with_stitch + stage_x_rot), 2)
+ np.power((stage_y_rot_with_stitch + stage_y_rot), 2)
)
<= pow((self.fov_circular / 2), 2)
<= pow((fov_circular / 2), 2)
)
)
def _prepare_setup(self):
yield from self.stubs.send_rpc_and_wait("rtx", "controller.clear_trajectory_generator")
yield from self.lamni_rotation(self.angle)
total_shift_x, total_shift_y = self._compute_total_shift()
total_shift_x, total_shift_y = self._compute_total_shift(
self.center_x,
self.center_y,
self.angle,
self.stitch_x,
self.stitch_y,
self.stitch_overlap,
self.shift_x,
self.shift_y,
self.fov_size,
)
yield from self.lamni_new_scan_center_interferometer(total_shift_x, total_shift_y)
# self._plot_target_pos()
if self.scan_type == "fly":
@@ -363,36 +385,73 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
step=self.step,
spiral_type=0,
center=False,
center_x=self.center_x,
center_y=self.center_y,
angle=self.angle,
stitch_x=self.stitch_x,
stitch_y=self.stitch_y,
stitch_overlap=self.stitch_overlap,
shift_x=self.shift_x,
shift_y=self.shift_y,
fov_size=self.fov_size,
fov_circular=self.fov_circular,
)
def _lamni_compute_stitch_center(self, xcount, ycount, angle_deg):
@staticmethod
def _lamni_compute_stitch_center(xcount, ycount, angle_deg, fov_size, stitch_overlap):
alpha = angle_deg / 180 * np.pi
stage_x = xcount * (self.fov_size[0] - self.stitch_overlap)
stage_y = ycount * (self.fov_size[1] - self.stitch_overlap)
stage_x = xcount * (fov_size[0] - stitch_overlap)
stage_y = ycount * (fov_size[1] - stitch_overlap)
x_rot = np.cos(alpha) * stage_x - np.sin(alpha) * stage_y
y_rot = np.sin(alpha) * stage_x + np.cos(alpha) * stage_y
return lamni_from_stage_coordinates(x_rot, y_rot)
def _compute_total_shift(self):
_shfitx, _shfity = self._lamni_compute_scan_center(self.center_x, self.center_y, self.angle)
x_stitch_shift, y_stitch_shift = self._lamni_compute_stitch_center(
self.stitch_x, self.stitch_y, self.angle
@staticmethod
def _compute_total_shift(
center_x, center_y, angle, stitch_x, stitch_y, stitch_overlap, shift_x, shift_y, fov_size
):
_shfitx, _shfity = LamNIFermatScan._lamni_compute_scan_center(center_x, center_y, angle)
x_stitch_shift, y_stitch_shift = LamNIFermatScan._lamni_compute_stitch_center(
stitch_x, stitch_y, angle, fov_size, stitch_overlap
)
logger.info(
f"Total shift [mm] {_shfitx+x_stitch_shift/1000+self.shift_x},"
f" {_shfity+y_stitch_shift/1000+self.shift_y}"
)
return (
_shfitx + x_stitch_shift / 1000 + self.shift_x,
_shfity + y_stitch_shift / 1000 + self.shift_y,
f"Total shift [mm] {_shfitx+x_stitch_shift/1000+shift_x},"
f" {_shfity+y_stitch_shift/1000+shift_y}"
)
return (_shfitx + x_stitch_shift / 1000 + shift_x, _shfity + y_stitch_shift / 1000 + shift_y)
@staticmethod
def get_lamni_fermat_spiral_pos(
self, m1_start, m1_stop, m2_start, m2_stop, step=1, spiral_type=0, center=False
m1_start,
m1_stop,
m2_start,
m2_stop,
step=1,
spiral_type=0,
center=False,
center_x=0.0,
center_y=0.0,
angle=0.0,
stitch_x=0,
stitch_y=0,
stitch_overlap=1,
shift_x=0.0,
shift_y=0.0,
fov_size=None,
fov_circular=0,
):
"""[summary]
Pure function (no device I/O) -- a @staticmethod rather than an
instance method (center_x/center_y/angle/stitch_x/stitch_y/
stitch_overlap/shift_x/shift_y/fov_size/fov_circular used to be read
off self.*) so it can also be called directly from client-side code
(tomo_parameters(), tomo_params.py) to predict the point count of a
not-yet-run scan, using the exact same algorithm the scan itself
will use -- including the rotated-stage and circular-FOV cropping
in _lamni_check_pos_in_fov_range_and_circ_fov().
Args:
m1_start (float): start position motor 1
m1_stop (float): end position motor 1
@@ -402,6 +461,7 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
spiral_type (float, optional): Angular offset in radians that determines the shape of the spiral.
A spiral with spiral_type=2 is the same as spiral_type=0. Defaults to 0.
center (bool, optional): Add a center point. Defaults to False.
fov_size (list): [fov_x, fov_y] used for the rotated-stage/circular-FOV crop.
Raises:
TypeError: [description]
@@ -423,7 +483,9 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
length_axis2 = np.abs(m2_stop - m2_start)
n_max = int(length_axis1 * length_axis2 * 3.2 / step / step)
total_shift_x, total_shift_y = self._compute_total_shift()
total_shift_x, total_shift_y = LamNIFermatScan._compute_total_shift(
center_x, center_y, angle, stitch_x, stitch_y, stitch_overlap, shift_x, shift_y, fov_size
)
for ii in range(start, n_max):
radius = step * 0.57 * np.sqrt(ii)
@@ -434,7 +496,9 @@ class LamNIFermatScan(AsyncFlyScanBase, LamNIMixin):
# continue
x = radius * np.sin(ii * phi)
y = radius * np.cos(ii * phi)
if self._lamni_check_pos_in_fov_range_and_circ_fov(x, y):
if LamNIFermatScan._lamni_check_pos_in_fov_range_and_circ_fov(
x, y, stitch_x, stitch_y, angle, fov_size, stitch_overlap, fov_circular
):
positions.extend([(x + total_shift_x * 1000, y + total_shift_y * 1000)])
# for testing we just shift by center_i and prepare also the setup to center_i
return np.array(positions)
+37 -15
View File
@@ -41,6 +41,13 @@ class FlomniFermatScan(AsyncFlyScanBase):
arg_input = {}
arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None}
# Minimum number of Fermat-spiral positions a scan is allowed to run
# with -- exposed as a class attribute (not just a literal inside
# _check_min_positions()) so client-side code (tomo_parameters(),
# tomo_params.py) can warn about a too-few-points configuration before
# it ever reaches the scan server, using the exact same threshold.
_MIN_POSITIONS = 20
def __init__(
self,
fovx: float,
@@ -148,9 +155,10 @@ class FlomniFermatScan(AsyncFlyScanBase):
self._check_min_positions()
def _check_min_positions(self):
if self.num_pos < 20:
if self.num_pos < self._MIN_POSITIONS:
raise ScanAbortion(
f"The number of positions must exceed 20. Currently: {self.num_pos}."
f"The number of positions must exceed {self._MIN_POSITIONS}. Currently:"
f" {self.num_pos}."
)
def _prepare_setup(self):
@@ -237,14 +245,33 @@ class FlomniFermatScan(AsyncFlyScanBase):
step=self.step,
spiral_type=0,
center=False,
cenx=self.cenx,
ceny=self.ceny,
zshift=self.zshift,
)
@staticmethod
def get_flomni_fermat_spiral_pos(
self, m1_start, m1_stop, m2_start, m2_stop, step=1, spiral_type=0, center=False
m1_start,
m1_stop,
m2_start,
m2_stop,
step=1,
spiral_type=0,
center=False,
cenx=0.0,
ceny=0.0,
zshift=0.0,
):
"""
Calculate positions for a Fermat spiral scan.
Pure function (no device I/O) -- a @staticmethod rather than an
instance method (cenx/ceny/zshift used to be read off self.*) so it
can also be called directly from client-side code (tomo_parameters(),
tomo_params.py) to predict the point count of a not-yet-run scan,
using the exact same algorithm the scan itself will use.
Args:
m1_start(float): start position in m1
m1_stop(float): stop position in m1
@@ -253,6 +280,9 @@ class FlomniFermatScan(AsyncFlyScanBase):
step(float): stepsize
spiral_type(int): 0 for traditional Fermat spiral
center(bool): whether to include the center position
cenx(float): center offset added to every x position
ceny(float): center offset added to every y position
zshift(float): z position for every point
Returns:
positions(array): positions
@@ -266,7 +296,7 @@ class FlomniFermatScan(AsyncFlyScanBase):
length_axis2 = np.abs(m2_stop - m2_start)
n_max = int(length_axis1 * length_axis2 * 3.2 / step / step)
z_pos = self.zshift
z_pos = zshift
for ii in range(start, n_max):
radius = step * 0.57 * np.sqrt(ii)
@@ -277,17 +307,9 @@ class FlomniFermatScan(AsyncFlyScanBase):
continue
x = radius * np.sin(ii * phi)
y = radius * np.cos(ii * phi)
positions.append([x + self.cenx, y + self.ceny, z_pos])
left_lower_corner = [
min(m1_start, m1_stop) + self.cenx,
min(m2_start, m2_stop) + self.ceny,
z_pos,
]
right_upper_corner = [
max(m1_start, m1_stop) + self.cenx,
max(m2_start, m2_stop) + self.ceny,
z_pos,
]
positions.append([x + cenx, y + ceny, z_pos])
left_lower_corner = [min(m1_start, m1_stop) + cenx, min(m2_start, m2_stop) + ceny, z_pos]
right_upper_corner = [max(m1_start, m1_stop) + cenx, max(m2_start, m2_stop) + ceny, z_pos]
positions.append(left_lower_corner)
positions.append(right_upper_corner)
return np.array(positions)
@@ -0,0 +1,134 @@
"""Tests for the Fermat-scan minimum-position warning: LamNI/Flomni's
_expected_fermat_position_count()/_fermat_min_positions() (lamni.py/
flomni.py), which predict a scan's point count before it ever reaches the
scan server, by calling the exact same algorithm the real scan classes use
(FlomniFermatScan.get_flomni_fermat_spiral_pos()/
LamNIFermatScan.get_lamni_fermat_spiral_pos(), both now pure @staticmethods
-- see csaxs_bec/scans/flomni_fermat_scan.py and LamNIFermatScan.py).
"""
import csaxs_bec.bec_ipython_client.plugins.LamNI.lamni as lamni_module
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni
from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan
class FakeClient:
"""Minimal in-memory stand-in for BEC's global-var store."""
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 _FakeRtx:
user_parameter = {"large_range_scan": True}
class _FakeDev:
rtx = _FakeRtx()
def __contains__(self, name):
return hasattr(self, name)
def __getitem__(self, name):
return getattr(self, name)
def make_flomni():
obj = object.__new__(Flomni)
obj.client = FakeClient()
return obj
def make_lamni():
# lamni_piezo_range_x/y's setter checks dev.rtx.user_parameter -- only
# ever defined (via the `if builtins.__dict__.get("bec") is not None`
# block at module import time) inside a real BEC session. Stand in a
# fake, same pattern as test_lamni_tomo_queue.py/test_tomo_queue_reacquire.py.
lamni_module.dev = _FakeDev()
obj = object.__new__(LamNI)
obj.client = FakeClient()
return obj
def test_flomni_min_positions_threshold_is_20():
assert FlomniFermatScan._MIN_POSITIONS == 20
def test_lamni_min_positions_threshold_is_20():
assert LamNIFermatScan._MIN_POSITIONS == 20
def test_flomni_expected_position_count_below_threshold():
flomni = make_flomni()
flomni.fovx = 5.0
flomni.fovy = 5.0
flomni.tomo_shellstep = 2.0
flomni.single_point_instead_of_fermat_scan = False
count = flomni._expected_fermat_position_count()
assert count < flomni._fermat_min_positions()
# matches calling the real scan-class algorithm directly
expected = FlomniFermatScan.get_flomni_fermat_spiral_pos(-2.5, 2.5, -2.5, 2.5, step=2.0)
assert count == len(expected)
def test_flomni_expected_position_count_above_threshold():
flomni = make_flomni()
flomni.fovx = 100.0
flomni.fovy = 80.0
flomni.tomo_shellstep = 1.0
flomni.single_point_instead_of_fermat_scan = False
count = flomni._expected_fermat_position_count()
assert count >= flomni._fermat_min_positions()
def test_lamni_expected_position_count_below_threshold():
lamni = make_lamni()
lamni.lamni_piezo_range_x = 5.0
lamni.lamni_piezo_range_y = 5.0
lamni.tomo_shellstep = 2.0
lamni.tomo_stitch_overlap = 0.2
lamni.tomo_circfov = 0.0
count = lamni._expected_fermat_position_count()
assert count < lamni._fermat_min_positions()
# matches calling the real scan-class algorithm directly, center tile, angle 0
expected = LamNIFermatScan.get_lamni_fermat_spiral_pos(
-2.5,
2.5,
-2.5,
2.5,
step=2.0,
angle=0.0,
stitch_x=0,
stitch_y=0,
stitch_overlap=0.2,
fov_size=[5.0, 5.0],
fov_circular=0.0,
)
assert count == len(expected)
def test_lamni_expected_position_count_above_threshold():
lamni = make_lamni()
lamni.lamni_piezo_range_x = 20.0
lamni.lamni_piezo_range_y = 20.0
lamni.tomo_shellstep = 1.0
lamni.tomo_stitch_overlap = 0.2
lamni.tomo_circfov = 0.0
count = lamni._expected_fermat_position_count()
assert count >= lamni._fermat_min_positions()
@@ -22,6 +22,7 @@ from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import (
LAMNI_QUEUE_PARAM_NAMES,
QUEUE_PARAM_NAMES,
SETUP_PROFILES,
_compute_fermat_positions_lamni,
_compute_type1,
_lamni_compute_type1,
_lamni_get_tomo_fov_offset,
@@ -29,6 +30,7 @@ from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import (
_lamni_set_tomo_fov_offset,
_requested_to_stepsize,
)
from csaxs_bec.scans.LamNIFermatScan import LamNIFermatScan
STEPSIZES = [10.0, 7.0, 25.0, 12.5]
@@ -165,3 +167,79 @@ def test_lamni_field_order_matches_tomo_parameters_cli():
"offsets",
"at_each_angle_hook",
]
def test_compute_fermat_positions_lamni_matches_scan_class_below_threshold():
params = {
"lamni_piezo_range_x": 5.0,
"lamni_piezo_range_y": 5.0,
"tomo_shellstep": 2.0,
"tomo_stitch_overlap": 0.2,
"tomo_circfov": 0.0,
"lamni_stitch_x": 0,
"lamni_stitch_y": 0,
}
count, min_positions = _compute_fermat_positions_lamni(params)
assert min_positions == LamNIFermatScan._MIN_POSITIONS
assert count < min_positions
expected = LamNIFermatScan.get_lamni_fermat_spiral_pos(
-2.5,
2.5,
-2.5,
2.5,
step=2.0,
angle=0.0,
stitch_x=0,
stitch_y=0,
stitch_overlap=0.2,
fov_size=[5.0, 5.0],
fov_circular=0.0,
)
assert count == len(expected)
def test_compute_fermat_positions_lamni_matches_scan_class_above_threshold():
params = {
"lamni_piezo_range_x": 20.0,
"lamni_piezo_range_y": 20.0,
"tomo_shellstep": 1.0,
"tomo_stitch_overlap": 0.2,
"tomo_circfov": 0.0,
"lamni_stitch_x": 0,
"lamni_stitch_y": 0,
}
count, min_positions = _compute_fermat_positions_lamni(params)
assert count >= min_positions
def test_compute_fermat_positions_lamni_uses_live_stitch_tile():
"""Unlike the CLI's tomo_parameters() (which always estimates the
center tile as a quick summary), the GUI has the actual stitch_x/y
fields right there, so it should reflect whichever tile is currently
being edited, not always stitch=0."""
base_params = {
"lamni_piezo_range_x": 20.0,
"lamni_piezo_range_y": 20.0,
"tomo_shellstep": 1.0,
"tomo_stitch_overlap": 0.2,
"tomo_circfov": 15.0,
"lamni_stitch_x": 0,
"lamni_stitch_y": 0,
}
shifted_params = dict(base_params, lamni_stitch_x=3, lamni_stitch_y=2)
count_center, _ = _compute_fermat_positions_lamni(base_params)
count_shifted, _ = _compute_fermat_positions_lamni(shifted_params)
# a large stitch offset combined with a tight circular FOV crop should
# shift the count -- if this ever becomes flaky because both happen to
# match, widen fov_circular/stitch further rather than removing the check.
assert count_center != count_shifted
def test_compute_fermat_positions_lamni_zero_step_is_safe():
count, min_positions = _compute_fermat_positions_lamni(
{"lamni_piezo_range_x": 5.0, "lamni_piezo_range_y": 5.0, "tomo_shellstep": 0.0}
)
assert count == 0
assert min_positions == LamNIFermatScan._MIN_POSITIONS
@@ -10,9 +10,11 @@ import pytest
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import (
_compute_fermat_positions_flomni,
_compute_type1,
_requested_to_stepsize,
)
from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan
STEPSIZES = [10.0, 7.0, 25.0, 12.5]
@@ -46,3 +48,28 @@ def _flomni_reference(stepsize):
N = int(180.0 / stepsize)
step = 180.0 / N
return N * 8, step, N
def test_compute_fermat_positions_flomni_matches_scan_class_below_threshold():
params = {"fovx": 5.0, "fovy": 5.0, "tomo_shellstep": 2.0}
count, min_positions = _compute_fermat_positions_flomni(params)
assert min_positions == FlomniFermatScan._MIN_POSITIONS
assert count < min_positions
expected = FlomniFermatScan.get_flomni_fermat_spiral_pos(-2.5, 2.5, -2.5, 2.5, step=2.0)
assert count == len(expected)
def test_compute_fermat_positions_flomni_matches_scan_class_above_threshold():
params = {"fovx": 100.0, "fovy": 80.0, "tomo_shellstep": 1.0}
count, min_positions = _compute_fermat_positions_flomni(params)
assert count >= min_positions
expected = FlomniFermatScan.get_flomni_fermat_spiral_pos(-50.0, 50.0, -40.0, 40.0, step=1.0)
assert count == len(expected)
def test_compute_fermat_positions_flomni_zero_step_is_safe():
count, min_positions = _compute_fermat_positions_flomni(
{"fovx": 5.0, "fovy": 5.0, "tomo_shellstep": 0.0}
)
assert count == 0
assert min_positions == FlomniFermatScan._MIN_POSITIONS