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

Fixes/flomni #260

Merged
holler merged 8 commits from fixes/flomni into main 2026-07-15 12:57:53 +02:00
Conversation 0 Commits 8 Files Changed 16
+981 -352
16 changed files with 981 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+349 -252
View File
@@ -27,6 +27,8 @@ from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import (
PtychoReconstructor,
TomoIDManager,
)
from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
# from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
# FlomniWebpageGenerator,
@@ -789,7 +791,8 @@ class FlomniSampleTransferMixin:
self.ensure_gripper_up()
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
self.flomni_modify_storage_non_interactive(100, 1, signal_name.get())
name, owner = unpack_desc(signal_name.get())
self.flomni_modify_storage_non_interactive(100, 1, name, owner=owner)
self.flomni_modify_storage_non_interactive(position, 0, "-")
def ftransfer_show_all(self):
@@ -846,9 +849,9 @@ class FlomniSampleTransferMixin:
self.ftransfer_controller_disable_mount_mode()
self.ensure_gripper_up()
sample_name = dev.flomni_samples.sample_in_gripper_name.get()
sample_name, sample_owner = unpack_desc(dev.flomni_samples.sample_in_gripper_name.get())
self.flomni_modify_storage_non_interactive(100, 0, "-")
self.flomni_modify_storage_non_interactive(position, 1, sample_name)
self.flomni_modify_storage_non_interactive(position, 1, sample_name, owner=sample_owner)
if position == 0:
self.ftransfer_flomni_stage_in()
@@ -859,7 +862,7 @@ class FlomniSampleTransferMixin:
Get the name of the sample currently in the given position.
"""
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
return signal_name.get()
return unpack_desc(signal_name.get())[0]
def ftransfer_sample_change(self, new_sample_position: int):
self.check_tray_in()
@@ -971,19 +974,24 @@ class FlomniSampleTransferMixin:
def ftransfer_modify_storage(self, position: int, used: int):
if used:
name = input("What's the name of this sample? ")
owner = input("Sample owner (optional): ")
else:
name = "-"
self.flomni_modify_storage_non_interactive(position, used, name)
owner = ""
self.flomni_modify_storage_non_interactive(position, used, name, owner=owner)
def flomni_modify_storage_non_interactive(self, position: int, used: int, name: str):
def flomni_modify_storage_non_interactive(
self, position: int, used: int, name: str, owner: str = ""
):
packed_name = pack_desc(name, owner)
if position == 100:
dev.flomni_samples.sample_in_gripper.set(used)
dev.flomni_samples.sample_in_gripper_name.set(name)
dev.flomni_samples.sample_in_gripper_name.set(packed_name)
else:
signal = getattr(dev.flomni_samples.sample_placed, f"sample{position}")
signal.set(used)
signal_name = getattr(dev.flomni_samples.sample_names, f"sample{position}")
signal_name.set(name)
signal_name.set(packed_name)
def check_position_is_valid(self, position: int):
if 0 <= position < 21:
@@ -1014,33 +1022,21 @@ class FlomniSampleTransferMixin:
"""
Hard abort of a running sample transfer routine on the Galil controller.
Stops the controller via dev.ftransy.stop(), which publishes a stop
request that the device server turns into motor.stop() and thus
XQ#STOP on the controller. #STOP halts the transfer thread (3),
aborts all motion (AB1) and clears mntprgs/mntmod. Afterwards the
controller is put back into positioning mode.
Delegates to GalilController.hard_abort_and_restore_positioning_mode()
-- the SAME method the GUI's hard-stop button calls directly (see
ConsoleButtonsWidget._on_abort()) -- so there is exactly one
implementation of this safety-critical stop sequence shared by both
the CLI (Ctrl+C) and GUI paths.
Deliberately does NOT call ensure_gripper_up(): after a mid-transfer
abort the gripper may be closed around a partially inserted sample,
so any recovery motion must be assessed and performed manually.
"""
print("Aborting sample transfer: stopping the controller routine.")
dev.ftransy.stop()
# The stop request is asynchronous (Redis -> device server). Wait
# until the transfer thread is actually halted before switching mode:
# #POSMODE refuses while mntprgs=1 and disable_mount_mode would raise.
timeout = 5
start = time.time()
while dev.ftransy.controller.is_thread_active(3):
if time.time() - start > timeout:
raise FlomniError(
"Transfer abort requested but the controller transfer routine (thread 3)"
f" did not stop within {timeout} s. Check the controller."
)
time.sleep(0.1)
# Ensure the controller is back in positioning mode. #STOP already
# clears mntmod, so this is mostly a verification step.
self.ftransfer_controller_disable_mount_mode()
try:
dev.ftransy.controller.hard_abort_and_restore_positioning_mode()
except GalilError as exc:
raise FlomniError(str(exc)) from exc
def ftransfer_confirm(self, step_name: str = ""):
confirm = int(float(dev.ftransy.controller.socket_put_and_receive("MG confirm").strip()))
@@ -2305,6 +2301,142 @@ class Flomni(
f.write(" ".join(map(str, x_vals)) + "\n")
f.write(" ".join(map(str, zeros)) + "\n")
@staticmethod
def _subtomo_angle_plan(subtomo_number, tomo_angle_range, tomo_angle_stepsize, start_angle=None):
"""
Pure angle-generation logic for one sub-tomogram of an equally
spaced 8-sub-tomogram tomography scan (tomo_type == 1). No device
I/O / progress side effects - kept separate from sub_tomo_scan()
so the angle math is directly unit-testable.
N/step are always computed against a fixed 180 degree span,
independent of tomo_angle_range: total projection count for a
given tomo_angle_stepsize is identical whether tomo_angle_range
is 180 or 360. In 180 mode all 8 sub-tomograms cover the same
[0,180) span (bit-reversal interlacing at step/8, unchanged from
the original scheme). In 360 mode each sub-tomogram instead
covers only a 180-degree span - never the full circle - split by
subtomo_number % 4 into a "low" half [0,180) (n%4 in (1,0)) and a
"high" half [180,360) (n%4 in (2,3)), each sub-tomogram in the
high half literally its low-half counterpart's own 180-degree
scan, shifted by +180 (per subtomo_number, using the SAME
phase_eighths value as 180 mode - not a different phase table).
Without this split, every sub-tomogram's own 360-degree sweep
would contain angle pairs exactly 180 degrees apart - redundant
tomographic information, since a measurement 180 degrees from one
already taken contributes no new information.
Reusing the same per-subtomo-number phase_eighths value in both
modes (rather than reassigning phases by acquisition order within
each half) is what makes subtomo N carry the same
interlacing/refinement role in both modes - e.g. subtomo 2 always
represents the phase_eighths[2] tier, whether that's within
[0,180) (180 mode) or shifted to [180,360) (360 mode, since
subtomo 2 % 4 == 2, a "high" sub-tomogram) - instead of an
unrelated phase suddenly appearing under the same subtomo number
depending on tomo_angle_range. Since the low half uses
subtomo_numbers {1,4,5,8} (phase_eighths values {0,6,1,7}) and the
high half uses {2,3,6,7} (phase_eighths values {4,2,5,3}), these
two phase sets are complementary (their union is all 8 eighths,
with no overlap) purely as a property of the original
phase_eighths table's structure - not something this branch has
to separately re-derive. Once the high half is folded mod 180,
low and high together reconstruct the full 8-way, step/8 grid
with every position hit EXACTLY once: the combined 360-degree
angle set is mod-180-identical to what an equivalent 180-degree
scan would produce, at the same total projection count, with zero
redundant measurements anywhere.
Returns:
angles (np.ndarray): the N angles (degrees) for this sub-tomogram.
subtomo_offset (int): index of this sub-tomogram's first angle
within the combined progress numbering (0 unless resuming).
N (int): number of projections in this (and every) sub-tomogram.
step (float): degree spacing between consecutive points within
this sub-tomogram.
"""
explicit_start_angle = start_angle is not None
# N/step not guaranteed to be a whole/exact division of the
# configured tomo_angle_stepsize; N is the actual, integer number
# of projections per sub-tomogram, step is the step size that's
# ACTUALLY achievable while landing exactly on N evenly-spaced
# points across a 180 degree span. This corrected step - not the
# raw, configured tomo_angle_stepsize - is used for BOTH the
# per-point ramp AND the inter-sub-tomogram phase offsets below.
N = int(180.0 / tomo_angle_stepsize)
step = 180.0 / N
# Same phase_eighths table in both modes - see docstring above for
# why reusing it (rather than a separately-derived phase table for
# 360 mode) is what keeps subtomo N's role consistent between
# modes.
phase_eighths = {1: 0, 2: 4, 3: 2, 4: 6, 5: 1, 6: 5, 7: 3, 8: 7}
phase = step / 8.0 * phase_eighths[subtomo_number]
if tomo_angle_range == 180:
base = 0.0
forward = bool(subtomo_number % 2)
else:
mod4 = subtomo_number % 4
base = 0.0 if mod4 in (1, 0) else 180.0
forward = mod4 in (1, 2)
if not explicit_start_angle:
if forward:
# Low end of this sub-tomogram's angular phase (same
# convention regardless of direction - it's what makes the
# combined sub-tomograms interlace into one fine angular
# grid).
start_angle = base + phase
else:
# A reverse sweep must begin at the HIGH end of this
# sub-tomogram's 180-degree span and descend. The literal
# high end (base + 180) would itself be rejected by
# _tomo_scan_at_angle's own "angle < tomo_angle_range +
# 0.05" gate for every sub-tomogram whose phase is nonzero
# (it lands just past the range), so start one step below
# that instead - this is the angle that will actually be
# the first one accepted. Skipped when start_angle is given
# explicitly (i.e. resuming mid sub-tomogram), since then
# the value is already the literal current angle.
start_angle = base + 180.0 - step + phase
# Every sub-tomogram covers exactly N projections, matching
# subtomo_total_projections elsewhere in this class - generated by
# plain arithmetic at the exact (corrected) step size, with no
# clamping. This deliberately never generates the boundary point at
# start +/- 180: that point is silently rejected by
# _tomo_scan_at_angle's own range gate for every sub-tomogram whose
# phase is nonzero anyway, so generating it only ever produced an
# inconsistent extra projection for the phase==0 sub-tomogram while
# every other sub-tomogram was silently one projection short.
if forward:
angles = start_angle + np.arange(N) * step
else:
angles = start_angle - np.arange(N) * step
if not explicit_start_angle:
# normal operation: always start at zero
subtomo_offset = 0
else:
# Explicitly subtract base and phase before dividing, rather
# than relying on an algebraic shortcut: for some sub-tomograms
# (e.g. #2 in 180 mode, #3/#4 in 360 mode), phase/step is
# exactly 0.5, landing precisely on the float rounding
# tie-break boundary - floating-point noise there
# unpredictably rounds up or down, which previously gave the
# wrong offset (off by +1) in ~44% of resumes within subtomo 2
# specifically (verified by exhaustive sweep). Every other
# sub-tomogram's phase fraction is far enough from 0.5 that an
# algebraic shortcut never broke for them.
if forward:
subtomo_offset = round((start_angle - base - phase) / step)
else:
subtomo_offset = round(((base + 180.0 - step + phase) - start_angle) / step)
return angles, subtomo_offset, N, step
def sub_tomo_scan(self, subtomo_number, start_angle=None):
"""
Performs a sub tomogram scan.
@@ -2313,117 +2445,21 @@ class Flomni(
start_angle (float, optional): The start angle of the scan. Defaults to None.
"""
explicit_start_angle = start_angle is not None
if explicit_start_angle:
if start_angle is not None:
print(f"Sub tomo scan with start angle {start_angle} requested.")
# tomo_angle_range / tomo_angle_stepsize is not guaranteed to be a
# whole number (e.g. a "total number of projections" that isn't a
# multiple of 8 was configured). N is the actual, integer number of
# projections per sub-tomogram; step is the step size that's
# ACTUALLY achievable while landing exactly on N evenly-spaced
# points across tomo_angle_range. This corrected step - not the
# raw, configured tomo_angle_stepsize - is used for BOTH the
# per-point ramp AND the inter-sub-tomogram phase offsets below.
# Using the raw stepsize for the phase offsets while the ramp used
# the corrected one is what caused the combined/interlaced
# tomogram to have an inconsistent angular spacing whenever N
# wasn't already a whole number for the raw stepsize.
N = int(self.tomo_angle_range / self.tomo_angle_stepsize)
step = self.tomo_angle_range / N
# Phase offset (degrees) for this sub-tomogram's position in the
# bit-reversal interlacing order - needed below to correctly
# recover the loop index i when resuming with an explicit
# start_angle (see the i==0 block further down).
phase_eighths = {1: 0, 2: 4, 3: 2, 4: 6, 5: 1, 6: 5, 7: 3, 8: 7}
phase = step / 8.0 * phase_eighths[subtomo_number]
if start_angle is None:
if subtomo_number == 1:
start_angle = 0
elif subtomo_number == 2:
start_angle = step / 8.0 * 4
elif subtomo_number == 3:
start_angle = step / 8.0 * 2
elif subtomo_number == 4:
start_angle = step / 8.0 * 6
elif subtomo_number == 5:
start_angle = step / 8.0 * 1
elif subtomo_number == 6:
start_angle = step / 8.0 * 5
elif subtomo_number == 7:
start_angle = step / 8.0 * 3
elif subtomo_number == 8:
start_angle = step / 8.0 * 7
if not subtomo_number % 2: # even = reverse
# The table above gives the LOW end of this sub-tomogram's
# angular phase (same convention as the forward/odd
# sub-tomograms - it's what makes the combined 8 sub-tomograms
# interlace into one fine angular grid). A reverse sweep must
# begin at the HIGH end of that span and descend. The literal
# high end (phase + tomo_angle_range) would itself be rejected
# by _tomo_scan_at_angle's own "angle < tomo_angle_range + 0.05"
# gate for every sub-tomogram whose phase is nonzero (it lands
# just past the range), so start one step below that instead -
# this is the angle that will actually be the first one
# accepted. This step is skipped when start_angle is given
# explicitly (i.e. we are resuming mid sub-tomogram), since
# then the value is already the literal current angle.
start_angle = start_angle + self.tomo_angle_range - step
# _tomo_shift_angles (potential global variable)
_tomo_shift_angles = 0
# compute number of projections
start = start_angle + _tomo_shift_angles
# Every sub-tomogram covers exactly N projections, matching
# subtomo_total_projections elsewhere in this class - generated by
# plain arithmetic at the exact (corrected) step size, with no
# clamping. This deliberately never generates the boundary point at
# start +/- tomo_angle_range: that point is silently rejected by
# _tomo_scan_at_angle's own range gate for every sub-tomogram whose
# phase is nonzero anyway, so generating it only ever produced an
# inconsistent extra projection for the phase==0 sub-tomogram while
# every other sub-tomogram was silently one projection short.
if subtomo_number % 2: # odd = forward: low -> high
angles = start + np.arange(N) * step
else: # even = reverse: high -> low
angles = start - np.arange(N) * step
angles, subtomo_offset, N, step = self._subtomo_angle_plan(
subtomo_number, self.tomo_angle_range, self.tomo_angle_stepsize, start_angle=start_angle
)
for i, angle in enumerate(angles):
self.progress["subtomo"] = subtomo_number
# --- NEW LOGIC FOR OFFSET WHEN start_angle IS SPECIFIED ---
if i == 0:
if not explicit_start_angle:
# normal operation: always start at zero
self._subtomo_offset = 0
else:
# Explicitly subtract the phase before dividing, rather
# than relying on an algebraic shortcut: for subtomo 2,
# phase/step is exactly 0.5 (its phase_eighths value is
# 4), landing precisely on the float rounding tie-break
# boundary - floating-point noise there unpredictably
# rounds up or down, which previously gave the wrong
# offset (off by +1) in ~44% of resumes within subtomo 2
# specifically (verified by exhaustive sweep). Every
# other sub-tomogram's phase fraction is far enough from
# 0.5 that the old shortcut never broke for them.
if subtomo_number % 2: # odd = forward direction
self._subtomo_offset = round((start_angle - phase) / step)
else: # even = reverse direction
self._subtomo_offset = round(
((phase + self.tomo_angle_range - step) - start_angle) / step
)
# progress index must always increase
if i == 0:
self._subtomo_offset = subtomo_offset
self.progress["subtomo_projection"] = self._subtomo_offset + i
# ------------------------------------------------------------
# existing progress fields. N is already an int (by
# construction, see above), so total_projections = N * 8 is
@@ -2439,6 +2475,16 @@ class Flomni(
# finally do the scan at this angle
self._tomo_scan_at_angle(angle, subtomo_number)
def _subtomo_starts_near_zero(self, subtomo_number: int) -> bool:
"""True if this sub-tomogram's own natural sweep begins near angle 0
(i.e. its direction is forward and its half is the low [0,180) one).
Used to gate the zero_deg_reference_at_each_subtomo damage-tracking
shot: forcing that shot before a sub-tomogram that doesn't actually
start near 0 would mean a large, wasted detour."""
if self.tomo_angle_range == 180:
return bool(subtomo_number % 2)
return subtomo_number % 4 == 1
@staticmethod
def _retry_unless_flomni_error(exc: Exception, attempt: int) -> bool:
"""scan_repeat() exc_handler: retry any exception except FlomniError.
@@ -2627,123 +2673,149 @@ class Flomni(
self.progress["heartbeat"] = None
self.collect_empty_frames()
with scans.dataset_id_on_hold:
if self.tomo_type == 1:
# 8 equally spaced sub-tomograms
self.progress["tomo_type"] = "Equally spaced sub-tomograms"
for ii in range(subtomo_start, 9):
if start_angle is None and ii % 2 and self.zero_deg_reference_at_each_subtomo:
# Dedicated reference shot at exactly 0 degrees, taken
# every time the rotation passes back through 0 (i.e.
# at the start of every odd/forward sub-tomogram), for
# tracking radiation damage over the full tomogram.
# Skipped when resuming mid-sub-tomogram (start_angle
# given explicitly) since we're not actually passing
# through 0 deg at that moment.
self._tomo_scan_at_angle(0, ii)
self.sub_tomo_scan(ii, start_angle=start_angle)
start_angle = None
if self.zero_deg_reference_at_each_subtomo:
# Final reference shot at exactly 0 degrees once the whole
# tomogram is complete, giving a clean "before vs after"
# pair together with sub-tomogram 1's own first projection
# (angle 0) for radiation-damage comparison across the
# full acquisition.
self._tomo_scan_at_angle(0, 8)
elif self.tomo_type == 2:
# Golden ratio tomography
previous_subtomo_number = -1
if projection_number == None:
ii = 0
else:
ii = projection_number
while True:
angle, subtomo_number = self._golden(ii, self.golden_ratio_bunch_size, 180, 1)
if previous_subtomo_number != subtomo_number:
try:
with scans.dataset_id_on_hold:
if self.tomo_type == 1:
# 8 equally spaced sub-tomograms
self.progress["tomo_type"] = "Equally spaced sub-tomograms"
for ii in range(subtomo_start, 9):
if (
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
start_angle is None
and self._subtomo_starts_near_zero(ii)
and self.zero_deg_reference_at_each_subtomo
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
self.progress["tomo_type"] = "Golden ratio tomography"
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
self.progress["angle"] = angle
if self.golden_ratio_bunch_size > 0:
self.progress["subtomo_total_projections"] = self.golden_ratio_bunch_size
# Dedicated reference shot at exactly 0 degrees, taken
# every time the rotation passes back through 0, for
# tracking radiation damage over the full tomogram.
# In 180 mode that's the start of every odd/forward
# sub-tomogram; in 360 mode only sub-tomograms 1 and 5
# (the "low half, forward" group) actually start near
# 0 - sub-tomograms 2/3/6/7 operate entirely in the
# [180,360) half and forcing a detour to 0 before them
# would defeat the point of the 360-mode boustrophedon
# path. Skipped when resuming mid-sub-tomogram
# (start_angle given explicitly) since we're not
# actually passing through 0 deg at that moment.
self._tomo_scan_at_angle(0, ii)
self.sub_tomo_scan(ii, start_angle=start_angle)
start_angle = None
if self.zero_deg_reference_at_each_subtomo:
# Final reference shot at exactly 0 degrees once the whole
# tomogram is complete, giving a clean "before vs after"
# pair together with sub-tomogram 1's own first projection
# (angle 0) for radiation-damage comparison across the
# full acquisition.
self._tomo_scan_at_angle(0, 8)
elif self.tomo_type == 2:
# Golden ratio tomography
previous_subtomo_number = -1
if projection_number == None:
ii = 0
else:
ii = projection_number
while True:
angle, subtomo_number = self._golden(
ii, self.golden_ratio_bunch_size, 180, 1
)
if previous_subtomo_number != subtomo_number:
if (
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
self.progress["tomo_type"] = "Golden ratio tomography"
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
self.progress["angle"] = angle
if self.golden_ratio_bunch_size > 0:
self.progress["subtomo_total_projections"] = (
self.golden_ratio_bunch_size
)
self.progress["subtomo_projection"] = (
ii - (subtomo_number - 1) * self.golden_ratio_bunch_size
)
else:
self.progress["subtomo_total_projections"] = 0
self.progress["subtomo_projection"] = 0
if self.golden_max_number_of_projections > 0:
self.progress["total_projections"] = (
self.golden_max_number_of_projections
)
else:
self.progress["total_projections"] = 0
self._tomo_scan_at_angle(angle, subtomo_number)
ii += 1
if (
ii > self.golden_max_number_of_projections
and self.golden_max_number_of_projections > 0
):
print(
f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
)
break
elif self.tomo_type == 3:
# Equally spaced tomography, golden ratio starting angle
previous_subtomo_number = -1
if projection_number == None:
ii = 0
else:
ii = projection_number
while True:
angle, subtomo_number = self._golden_equally_spaced(
ii, int(180 / self.tomo_angle_stepsize), 180, 1, 0
)
if previous_subtomo_number != subtomo_number:
if (
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
self.progress["tomo_type"] = (
"Equally spaced tomography, golden ratio starting angle"
)
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
self.progress["angle"] = angle
self.progress["subtomo_total_projections"] = 180 / self.tomo_angle_stepsize
self.progress["subtomo_projection"] = (
ii - (subtomo_number - 1) * self.golden_ratio_bunch_size
ii - (subtomo_number - 1) * self.progress["subtomo_total_projections"]
)
else:
self.progress["subtomo_total_projections"] = 0
self.progress["subtomo_projection"] = 0
if self.golden_max_number_of_projections > 0:
self.progress["total_projections"] = self.golden_max_number_of_projections
else:
self.progress["total_projections"] = 0
self._tomo_scan_at_angle(angle, subtomo_number)
ii += 1
if (
ii > self.golden_max_number_of_projections
and self.golden_max_number_of_projections > 0
):
print(
f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
)
break
elif self.tomo_type == 3:
# Equally spaced tomography, golden ratio starting angle
previous_subtomo_number = -1
if projection_number == None:
ii = 0
else:
ii = projection_number
while True:
angle, subtomo_number = self._golden_equally_spaced(
ii, int(180 / self.tomo_angle_stepsize), 180, 1, 0
)
if previous_subtomo_number != subtomo_number:
if self.golden_max_number_of_projections > 0:
self.progress["total_projections"] = (
self.golden_max_number_of_projections
)
else:
self.progress["total_projections"] = 0
self._tomo_scan_at_angle(angle, subtomo_number)
ii += 1
if (
subtomo_number % 2 == 1
and ii > 10
and self.golden_projections_at_0_deg_for_damage_estimation == 1
ii > self.golden_max_number_of_projections
and self.golden_max_number_of_projections > 0
):
self._tomo_scan_at_angle(0, subtomo_number)
previous_subtomo_number = subtomo_number
self.progress["tomo_type"] = (
"Equally spaced tomography, golden ratio starting angle"
)
self.progress["subtomo"] = subtomo_number
self.progress["projection"] = ii
self.progress["angle"] = angle
self.progress["subtomo_total_projections"] = 180 / self.tomo_angle_stepsize
self.progress["subtomo_projection"] = (
ii - (subtomo_number - 1) * self.progress["subtomo_total_projections"]
)
if self.golden_max_number_of_projections > 0:
self.progress["total_projections"] = self.golden_max_number_of_projections
else:
self.progress["total_projections"] = 0
self._tomo_scan_at_angle(angle, subtomo_number)
ii += 1
if (
ii > self.golden_max_number_of_projections
and self.golden_max_number_of_projections > 0
):
print(
f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
)
break
else:
raise FlomniError("undefined tomo type")
print(
f"Golden ratio tomography stopped automatically after the requested {self.golden_max_number_of_projections} projections"
)
break
else:
raise FlomniError("undefined tomo type")
finally:
# Cleared on every exit path (normal completion, exception, or
# interrupt/abort) - not just on the next scan's start (line
# ~2664) - so the GUI's busy-detectors (TomoParamsWidget,
# FlomniWebpageGenerator) see this immediately on their next
# poll instead of waiting out their heartbeat-staleness timeout
# (120s / 90s) after a tomogram that's actually already done.
self.progress["heartbeat"] = None
self.progress["projection"] = self.progress["total_projections"]
self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
@@ -3537,11 +3609,16 @@ class Flomni(
def _tomo_type1_actual_grid(self) -> tuple[int, float, int]:
"""Compute the actual (achievable) tomo_type==1 grid from the
currently stored self.tomo_angle_stepsize -- the SAME way
sub_tomo_scan() does it. Returns (N, step, total_projections):
sub_tomo_scan()/_subtomo_angle_plan() does it. Returns (N, step,
total_projections):
N: integer number of projections per sub-tomogram
step: the achievable per-projection angular step (range / N) --
this is what the scan actually runs at, NOT
self.tomo_angle_stepsize itself
step: the achievable per-projection angular step within a
sub-tomogram's own 180-degree span -- this is what the scan
actually runs at, NOT self.tomo_angle_stepsize itself. N/step
are always computed against a fixed 180 degrees, independent
of tomo_angle_range: total projection count for a given
tomo_angle_stepsize is the same whether tomo_angle_range is
180 or 360 (see _subtomo_angle_plan()'s docstring).
total_projections: N * 8
self.tomo_angle_stepsize is stored as a raw, uncorrected value
@@ -3554,8 +3631,8 @@ class Flomni(
the displayed "angular step within sub-tomogram" to silently differ
from the angle the scan was actually acquiring at.
"""
N = int(self.tomo_angle_range / self.tomo_angle_stepsize)
step = self.tomo_angle_range / N
N = int(180.0 / self.tomo_angle_stepsize)
step = 180.0 / N
return N, step, N * 8
def tomo_parameters(self):
@@ -3592,10 +3669,18 @@ class Flomni(
_, achievable_step, total_projections = self._tomo_type1_actual_grid()
print(f"Total number of projections: {total_projections}")
print(f"Angular step within sub-tomogram: {achievable_step:.3f} degrees")
# Always 180/total_projections, never tomo_angle_range/total_projections:
# in 360 mode the low/high halves are complementary (see
# _subtomo_angle_plan()'s docstring), so the combined tomogram's
# true resolution matches 180 mode exactly for the same total
# count - it is not tomo_angle_range/total_projections, which
# would (wrongly) show a different, coarser number for 360 mode.
print(
"Angular step of the final (combined) tomogram:"
f" {self.tomo_angle_range / total_projections:.3f} degrees"
f" {180.0 / total_projections:.3f} degrees"
)
if self.tomo_angle_range == 360:
print("There are no duplicate angles.")
print(
"0-deg reference shots (odd sub-tomo start + end) ="
f" {self.zero_deg_reference_at_each_subtomo}"
@@ -3695,7 +3780,10 @@ class Flomni(
tomo_numberofprojections = self._get_val(
"Total number of projections", current_total, int
)
self.tomo_angle_stepsize = (self.tomo_angle_range / tomo_numberofprojections) * 8
# N/step (and therefore total projections) are always
# computed against a fixed 180 degrees, independent of
# tomo_angle_range -- see _subtomo_angle_plan()'s docstring.
self.tomo_angle_stepsize = (180.0 / tomo_numberofprojections) * 8
# Now report what was ACTUALLY achieved, via the same helper
# sub_tomo_scan() effectively uses -- not the raw value just
@@ -3716,10 +3804,15 @@ class Flomni(
print(
f"The angular step within a sub-tomogram will be {achievable_step:.3f} degrees"
)
# Always 180/actual_total - see the same line in
# tomo_parameters() for why tomo_angle_range/actual_total
# would be wrong here.
print(
"The angular step of the final (combined) tomogram will be"
f" {self.tomo_angle_range / actual_total:.3f} degrees"
f" {180.0 / actual_total:.3f} degrees"
)
if self.tomo_angle_range == 360:
print("There are no duplicate angles.")
self.zero_deg_reference_at_each_subtomo = bool(
self._get_val(
"Take 0-deg reference shots (start of each odd sub-tomo + end) for"
@@ -4337,15 +4430,19 @@ class Flomni(
stitching = f"{self.stitch_x:.0f}/{self.stitch_y:.0f}"
dataset_id = str(self.client.queue.next_dataset_number)
account = bec.active_account
# Same grid sub_tomo_scan() actually uses -- see
# _tomo_type1_actual_grid()'s docstring for why this can't just
# recompute int((tomo_angle_range/tomo_angle_stepsize)*8) locally.
_, _, tomo_type1_total_projections = self._tomo_type1_actual_grid()
content = [
f"{'Sample Name:':<{padding}}{self.sample_name:>{padding}}\n",
f"{'Measurement ID:':<{padding}}{str(self.tomo_id):>{padding}}\n",
f"{'Dataset ID:':<{padding}}{dataset_id:>{padding}}\n",
f"{'Sample Info:':<{padding}}{'Sample Info':>{padding}}\n",
f"{'e-account:':<{padding}}{str(account):>{padding}}\n",
f"{'Number of projections:':<{padding}}{int((self.tomo_angle_range / self.tomo_angle_stepsize) * 8):>{padding}}\n",
f"{'Number of projections:':<{padding}}{tomo_type1_total_projections:>{padding}}\n",
f"{'First scan number:':<{padding}}{self.client.queue.next_scan_number:>{padding}}\n",
f"{'Last scan number approx.:':<{padding}}{self.client.queue.next_scan_number + int((self.tomo_angle_range / self.tomo_angle_stepsize) * 8) + 10:>{padding}}\n",
f"{'Last scan number approx.:':<{padding}}{self.client.queue.next_scan_number + tomo_type1_total_projections + 10:>{padding}}\n",
f"{'Current photon energy:':<{padding}}To be implemented\n",
# f"{'Current photon energy:':<{padding}}{dev.mokev.read()['mokev']['value']:>{padding}.4f}\n",
f"{'Exposure time:':<{padding}}{self.tomo_countingtime:>{padding}.2f}\n",
csaxs_bec/bec_ipython_client/plugins/flomni/flomni_optics_mixin.py
+16 -4
View File
@@ -34,7 +34,9 @@ class FlomniOpticsMixin:
if "fttrx1" not in dev:
if not self.OMNYTools.yesno(
"Device 'fttrx1' does not exist in the current config, so it "
"can't be moved out of the way. Continue anyway?"
"can't be moved out of the way. Continue anyway? WARNING: "
"this could cause collisions when operating on the real "
"beamline."
):
print("Aborting feye_out(): fttrx1 not moved.")
return
@@ -44,9 +46,19 @@ class FlomniOpticsMixin:
def feye_in(self):
bec.queue.next_dataset_number += 1
fttrx_out = self._get_user_param_safe("feyex", "fttrx_out")
umv(dev.fttrx1, fttrx_out)
if "fttrx1" not in dev:
if not self.OMNYTools.yesno(
"Device 'fttrx1' does not exist in the current config, so it "
"can't be moved into position. Continue anyway? WARNING: "
"this could cause collisions when operating on the real "
"beamline."
):
print("Aborting feye_in(): fttrx1 not moved.")
return
else:
fttrx_out = self._get_user_param_safe("feyex", "fttrx_out")
umv(dev.fttrx1, fttrx_out)
feyex_in = self._get_user_param_safe("feyex", "in")
feyey_in = self._get_user_param_safe("feyey", "in")
csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py
+14 -4
View File
@@ -615,7 +615,13 @@ class WebpageGeneratorBase:
counts for queued jobs without hardcoding a
per-instrument formula in JS. Recognised "kind"s:
"equally_spaced_grid" (params: n_subtomos)
total = floor(angle_range / angle_stepsize) * n_subtomos
total = floor(180 / angle_stepsize) * n_subtomos
(independent of angle_range - see
Flomni._subtomo_angle_plan()'s docstring for
why: in 360-degree mode, angle_range only
splits the n_subtomos into complementary
low/high halves, it does not change the
total count)
"golden_capped"
total = golden_max_number_of_projections
"""
@@ -2842,9 +2848,13 @@ function calcProjections(params){{
return (gmax!=null&&gmax>0)?gmax:null;
}}
if(def.kind==='equally_spaced_grid'){{
const range=params.tomo_angle_range, step=params.tomo_angle_stepsize;
if(range>0&&step>0){{
const N=Math.floor(range/step);
// Always floor(180/step), never floor(angle_range/step): in 360-degree
// mode angle_range only splits the sub-tomograms into complementary
// low/high halves, it does not change the total projection count -
// see Flomni._subtomo_angle_plan()'s docstring.
const step=params.tomo_angle_stepsize;
if(step>0){{
const N=Math.floor(180/step);
return N*(def.n_subtomos||1);
}}
}}
csaxs_bec/bec_ipython_client/plugins/flomni/gui_tools.py
+15 -2
View File
@@ -131,9 +131,22 @@ class flomniGuiTools:
else:
print("Cannot open camera_overview. Device does not exist.")
# Confirm/abort console, docked below the cameras.
# Confirm/abort console, docked below the cameras. The hard-stop
# button directly calls dev.ftransy.controller
# .hard_abort_and_restore_positioning_mode() (the same Galil
# hard-stop ftransfer_abort() uses) -- see ConsoleButtonsWidget's
# docstring for why this replaced a blind stop-all-devices
# broadcast. foptx sits on a physically separate Galil
# controller (the optics stage board, not the transfer stage
# board) so it needs its own stop_all_axes() call -- also
# explained in ConsoleButtonsWidget's docstring.
self.console = self.gui.flomni.new(
"ConsoleButtonsWidget", object_name="console", where="bottom"
"ConsoleButtonsWidget",
object_name="console",
where="bottom",
hard_stop_device_name="ftransy",
extra_hard_stop_device_name="foptx",
hard_stop_label="Flomni Motion Stop",
)
# set_layout_ratios uses relative weights, not pixels -- there is
# no width/height kwarg on dock_area.new(). [5, 1] gives the
csaxs_bec/bec_ipython_client/plugins/omny/omny_general_tools.py
+5 -4
View File
@@ -134,10 +134,11 @@ class OMNYTools:
bool: True for "yes", False for "no".
Note on abort: the widget's ABORT button does not write a response
to poll for -- it sends a real SIGINT directly to this process (see
ConsoleButtonsWidget._on_abort), so pressing it raises
KeyboardInterrupt here exactly as a console Ctrl+C would, and
propagates normally out of this method without any special-casing.
to poll for -- it directly calls a configured hard motion stop and
sends a SIGINT to this process (see ConsoleButtonsWidget._on_abort),
so pressing it raises KeyboardInterrupt here exactly as a console
Ctrl+C would, and propagates normally out of this method without any
special-casing.
"""
if autoconfirm and default == "y":
self.printgreen(message + " Automatically confirming default: yes")
csaxs_bec/bec_widgets/widgets/console_buttons/console_buttons.py
+126 -46
View File
@@ -4,11 +4,8 @@ import os
import signal
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import VariableMessage
from bec_widgets import BECWidget, SafeProperty, SafeSlot
from bec_widgets.utils.rpc_decorator import rpc_timeout
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidget
logger = bec_logger.logger
@@ -17,36 +14,69 @@ logger = bec_logger.logger
class ConsoleButtonsWidget(BECWidget, QWidget):
"""
Small Yes / No / Abort control widget, intended as a GUI replacement for
console prompts (e.g. ``OMNYTools.yesno()``) and as a general-purpose
emergency-stop control.
console prompts (e.g. ``OMNYTools.yesno()``) and as a hard motion-stop
control for a Galil-controlled beamline (flomni/OMNY/LamNI).
- Yes / No: set ``response`` to "yes" / "no". A blocking CLI script can
poll ``gui.<name>.response`` and reset it via ``clear_response()``.
- Abort: sends a real SIGINT to the BEC IPython client process (the
parent of the GUI server process), equivalent to pressing Ctrl+C in
the console. This works even if the client is blocked inside a motor
move or other long call, since it is a real OS signal rather than a
polled flag. The SIGINT is only sent if the parent process actually
looks like a BEC client -- it does not when the widget is opened
standalone from the launcher menu. In addition, 500 ms later a device
stop request is published to the device server (same mechanism as the
PositionerBox stop button): by default for ALL devices, so the widget
acts as a generic emergency stop from any context, even if the client
process is hung or dead.
- Abort: calls ``queue.request_scan_abortion()`` (a safe no-op if
nothing is queued, but registers a stop_id so the device server
suppresses the resulting error if something else happens to be mid a
queue-tracked instruction at the same time -- this is what previously
let an unrelated, un-suppressed device error kill the scan worker
thread and require a full BEC restart), then sends a real SIGINT to
the BEC IPython client process (the parent of the GUI server
process), then directly calls ``hard_stop_device_name``'s
``.controller.hard_abort_and_restore_positioning_mode()`` -- the same
Galil hard-stop (XQ#STOP,1) the CLI's ``ftransfer_abort()`` uses --
and, if configured, ``extra_hard_stop_device_name``'s
``.controller.stop_all_axes()``.
SIGINT is sent BEFORE the hard stop, not after: if a client-side loop
is actively polling this same transfer (e.g. ftransfer_get_sample's
``while True: ... if not in_progress: break``), and the hard stop's
own controller-level clear of ``mntprgs`` lands before that process
notices the interrupt, the loop would see "not in progress" and take
that as normal completion -- silently continuing into
``ensure_gripper_up()``, which must not run mid-transfer. Sending
SIGINT first (an os.kill() call, effectively instant) gives the
target process's own ``except KeyboardInterrupt: ftransfer_abort();
raise FlomniError(...)`` handler a head start over the hard stop's
own multi-step sequence (which itself takes hundreds of ms due to
its internal polling/settling waits) -- without adding any
artificial delay of its own, so the actual motion stop is still
effectively immediate.
``hard_stop_device_name`` names the device whose ``.controller`` the
hard stop is called on (e.g. ``"ftransy"`` for flomni). Without a device
*name* configured, the button is disabled rather than silently doing
nothing. A configured name that doesn't actually resolve to a device
(e.g. missing from the current session's config) is not checked ahead
of time -- it's re-resolved at click time in ``_on_abort()``, which
already logs and no-ops on failure -- so the button being enabled is
not a guarantee the device exists, only that a target was configured.
``extra_hard_stop_device_name`` optionally names a second device on a
physically separate Galil controller (e.g. ``"foptx"`` for flomni's
optics stage, a different board/socket port from the transfer stage
``ftransy`` sits on -- one abort press only halts axes on ``ftransy``'s
controller otherwise). Its ``.controller.stop_all_axes()`` is called
instead of ``hard_abort_and_restore_positioning_mode()``: the latter's
``#POSMODE``/``mntmod`` handling is specific to the sample-transfer/
mount program that only runs on the transfer controller, so
``stop_all_axes()`` (a plain ``XQ#STOP,1``) is the correct generic
"halt everything on this controller" primitive for any other board.
Entirely optional -- unset means only ``hard_stop_device_name`` is
stopped, as before.
"""
USER_ACCESS = ["message", "message.setter", "response", "clear_response"]
PLUGIN = True
def __init__(self, parent=None, **kwargs):
# Devices for which a backup stop request is published when ABORT is
# pressed. An empty list (default) means "stop ALL devices" -- the
# same device-server path BEC uses on scan abort -- which makes the
# widget a generic emergency stop, e.g. when opened standalone from
# the launcher menu. Stop-all also covers the flomni sample transfer:
# XQ#STOP via ftransy is controller-wide and halts the #GRGET/#GRPUT
# thread. Pass an explicit list to restrict the stop.
self._backup_stop_devices = list(kwargs.pop("backup_stop_devices", []))
self._hard_stop_device_name = kwargs.pop("hard_stop_device_name", None)
self._extra_hard_stop_device_name = kwargs.pop("extra_hard_stop_device_name", None)
self._hard_stop_label = kwargs.pop("hard_stop_label", "Motion Stop")
super().__init__(parent=parent, **kwargs)
self._response = ""
# Captured once at construction time: the GUI server process is a
@@ -57,6 +87,23 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
self._client_pid = os.getppid()
self._init_ui()
def _hard_stop_available(self) -> bool:
"""True if a hard_stop_device_name was configured for this widget.
Deliberately does not check whether the name actually resolves to a
live, enabled device: that live check (a getattr(self.dev, name)
device-manager lookup) proved unreliable in practice -- a
confirmed-present, confirmed-enabled device could still evaluate as
unavailable here, permanently disabling the button for the widget's
lifetime with no way to recover short of recreating the dock. Since
whatever device is named is re-resolved with its own try/except in
_on_abort() anyway, a stale or wrong name still fails safely
(logged, no-op) instead of silently doing nothing -- so skipping the
redundant, unreliable pre-check here only removes a false-negative
failure mode, not real safety.
"""
return bool(self._hard_stop_device_name)
def _init_ui(self):
layout = QVBoxLayout(self)
@@ -81,11 +128,19 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
self.abort_button.clicked.connect(self._on_abort)
# Start with Yes/No greyed out: with no message there is nothing to
# respond to, so there should be nothing clickable. ABORT is left
# always enabled -- it's an emergency stop and must work at any time,
# message or not.
# respond to, so there should be nothing clickable.
self._set_yesno_enabled(False)
# A dead-but-clickable "ABORT" button (no configured hard-stop
# target) is worse than an obviously disabled one -- disable it
# rather than silently doing nothing useful when pressed.
if self._hard_stop_available():
self.abort_button.setText(self._hard_stop_label)
self.abort_button.setEnabled(True)
else:
self.abort_button.setText("Motion Stop (not active)")
self.abort_button.setEnabled(False)
def _set_yesno_enabled(self, enabled: bool):
self.yes_button.setEnabled(enabled)
self.no_button.setEnabled(enabled)
@@ -115,29 +170,54 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
@SafeSlot()
def _on_abort(self):
# 1) Coordinated abort first: a safe no-op if nothing is queued: when
# something IS queue-tracked, this registers a real stop_id so the
# device server suppresses the resulting error instead of killing
# the scan worker thread (see class docstring).
try:
self.queue.request_scan_abortion()
except Exception:
logger.exception("ConsoleButtonsWidget: request_scan_abortion() failed")
# 2) SIGINT before the hard stop -- gives a client-side polling loop
# (if this is a self-abort) a head start to notice the interrupt
# before the hard stop's own controller-level state change could
# let it exit "cleanly" instead. See class docstring for why the
# order matters here. Only sent if the parent process actually
# looks like a BEC client -- it does not when the widget is
# opened standalone from the launcher menu.
if self._client_is_bec_process():
logger.warning(f"ConsoleButtonsWidget: sending SIGINT to client pid {self._client_pid}")
os.kill(self._client_pid, signal.SIGINT)
else:
logger.warning(
"ConsoleButtonsWidget: parent process does not look like a BEC client;"
" skipping SIGINT and only sending the device stop request."
)
# Backup: direct device stop via the device server, independent of
# the client process. Delayed so the SIGINT-triggered abort handler
# in the client (which still sees mntprgs=1 and aborts in a
# controlled way) wins the race: if the stop landed first, #STOP
# would clear mntprgs and the client transfer loop would exit
# "cleanly" into ensure_gripper_up, which must not happen
# mid-transfer.
QTimer.singleShot(500, self._send_backup_stop)
@SafeSlot()
def _send_backup_stop(self):
"""Publish a stop request for the configured devices to the device server."""
devices = self._backup_stop_devices
logger.warning(f"ConsoleButtonsWidget: sending backup stop request for {devices}")
self.client.connector.send(MessageEndpoints.stop_devices(), VariableMessage(value=devices))
# 3) Hard motion stop(s) -- not delayed by anything above: sending
# SIGINT is a near-instant os.kill() call, so these still run
# effectively immediately. The two devices are independent
# controllers (see class docstring): each is attempted and
# logged on its own, so a failure on one doesn't skip the other.
if self._hard_stop_device_name:
self._stop_device_controller(
self._hard_stop_device_name, "hard_abort_and_restore_positioning_mode"
)
if self._extra_hard_stop_device_name:
self._stop_device_controller(self._extra_hard_stop_device_name, "stop_all_axes")
def _stop_device_controller(self, device_name: str, method_name: str) -> None:
"""Resolve device_name and call method_name() on its .controller,
logging and swallowing any failure -- so a missing device or a
failed stop on one controller can't prevent an already-issued stop
on another, or crash the rest of _on_abort()."""
try:
device = getattr(self.dev, device_name, None)
except Exception:
device = None
if device is None:
return
logger.warning(f"ConsoleButtonsWidget: hard-stopping {device_name} ({method_name})")
try:
getattr(device.controller, method_name)()
except Exception:
logger.exception(f"ConsoleButtonsWidget: hard motion stop of {device_name} failed")
@SafeProperty(str)
def message(self):
csaxs_bec/bec_widgets/widgets/sample_storage/sample_storage.py
+52 -25
View File
@@ -62,6 +62,8 @@ from qtpy.QtWidgets import (
QWidget,
)
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
logger = bec_logger.logger
# ── constants ────────────────────────────────────────────────────────────────
@@ -107,6 +109,7 @@ class _SlotCell(QFrame):
self._owner = owner
self._occupied = False
self._name = EMPTY_NAME
self._sample_owner = ""
self.setFrameShape(QFrame.Shape.Box)
self.setLineWidth(1)
@@ -126,8 +129,14 @@ class _SlotCell(QFrame):
self._lbl_name.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._lbl_name.setWordWrap(True)
self._lbl_owner = QLabel("")
self._lbl_owner.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._lbl_owner.setStyleSheet("color: #888888; font-size: 10px;")
self._lbl_owner.setWordWrap(True)
layout.addWidget(self._lbl_num)
layout.addWidget(self._lbl_name, stretch=1)
layout.addWidget(self._lbl_owner)
self.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
self.customContextMenuRequested.connect(self._show_menu)
@@ -141,16 +150,19 @@ class _SlotCell(QFrame):
return "Gripper (100)"
return str(self._slot)
def set_state(self, occupied: bool, name: str) -> None:
def set_state(self, occupied: bool, name: str, owner: str = "") -> None:
"""Update the cell's displayed state (called by the poll/refresh)."""
self._occupied = occupied
self._name = name if name else EMPTY_NAME
self._sample_owner = owner
if occupied:
self._lbl_name.setText(self._name)
self._lbl_name.setStyleSheet(f"color: {COLOR_OCCUPIED};")
self._lbl_owner.setText(f"owner: {owner}" if owner else "")
else:
self._lbl_name.setText(EMPTY_LABEL)
self._lbl_name.setStyleSheet(f"color: {COLOR_EMPTY}; font-style: italic;")
self._lbl_owner.setText("")
# ── context menu ──────────────────────────────────────────────────────────
@@ -161,7 +173,7 @@ class _SlotCell(QFrame):
act_clear = menu.addAction("Clear (set empty)")
chosen = menu.exec_(self.mapToGlobal(pos))
if chosen == act_rename:
self._owner.action_change_name(self._slot, self._name)
self._owner.action_change_name(self._slot, self._name, self._sample_owner)
elif chosen == act_clear:
self._owner.action_clear(self._slot)
else:
@@ -198,8 +210,8 @@ class SampleStorageWidget(BECWidget, QWidget):
super().__init__(parent=parent, **kwargs)
self.get_bec_shortcuts()
self._cells: dict[int, _SlotCell] = {}
# last-seen (occupied, name) per slot, for change-detection in refresh()
self._last_state: dict[int, tuple[bool, str]] = {}
# last-seen (occupied, name, owner) per slot, for change-detection in refresh()
self._last_state: dict[int, tuple[bool, str, str]] = {}
self._flomni_available = self._check_flomni_available()
self._build_ui()
self._poll_timer = QTimer(self)
@@ -229,9 +241,9 @@ class SampleStorageWidget(BECWidget, QWidget):
"""The flomni_samples device (single source of truth)."""
return self.dev.flomni_samples
def _read_all_slots(self) -> dict[int, tuple[bool, str]]:
"""Return {slot: (occupied, name)} for every slot 0–20 and the gripper
from a SINGLE bulk ``dev.flomni_samples.read()`` round-trip.
def _read_all_slots(self) -> dict[int, tuple[bool, str, str]]:
"""Return {slot: (occupied, name, owner)} for every slot 0–20 and the
gripper from a SINGLE bulk ``dev.flomni_samples.read()`` round-trip.
The per-slot accessors (``is_sample_slot_used`` +
``sample_names.sample{N}.get()``) would be ~44 blocking device
@@ -242,9 +254,10 @@ class SampleStorageWidget(BECWidget, QWidget):
``flomni_samples_sample_names_sample{N}`` /
``flomni_samples_sample_in_gripper`` /
``flomni_samples_sample_in_gripper_name``), the same keys the CLI
iterated over in ftransfer_sample_change.
iterated over in ftransfer_sample_change. The raw name signal packs
name+owner (see ``sample_desc_codec``), unpacked here.
"""
result: dict[int, tuple[bool, str]] = {}
result: dict[int, tuple[bool, str, str]] = {}
try:
data = self._samples.read()
except Exception as exc:
@@ -260,27 +273,30 @@ class SampleStorageWidget(BECWidget, QWidget):
for slot in (STAGE_SLOT, *STORAGE_SLOTS):
used = _val(f"flomni_samples_sample_placed_sample{slot}", 0)
name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME)
result[slot] = (bool(used), name if name else EMPTY_NAME)
raw_name = _val(f"flomni_samples_sample_names_sample{slot}", EMPTY_NAME)
name, owner = unpack_desc(raw_name if raw_name else EMPTY_NAME)
result[slot] = (bool(used), name if name else EMPTY_NAME, owner)
g_used = _val("flomni_samples_sample_in_gripper", 0)
g_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME)
result[GRIPPER_SLOT] = (bool(g_used), g_name if g_name else EMPTY_NAME)
g_raw_name = _val("flomni_samples_sample_in_gripper_name", EMPTY_NAME)
g_name, g_owner = unpack_desc(g_raw_name if g_raw_name else EMPTY_NAME)
result[GRIPPER_SLOT] = (bool(g_used), g_name if g_name else EMPTY_NAME, g_owner)
return result
def _write_slot(self, slot: int, used: int, name: str) -> bool:
def _write_slot(self, slot: int, used: int, name: str, owner: str = "") -> bool:
"""
Write (used, name) to a storage slot 0–20 or the gripper, mirroring
``Flomni.flomni_modify_storage_non_interactive()`` exactly.
Write (used, name, owner) to a storage slot 0–20 or the gripper,
mirroring ``Flomni.flomni_modify_storage_non_interactive()`` exactly.
"""
try:
packed_name = pack_desc(name, owner)
if slot == GRIPPER_SLOT:
self._samples.sample_in_gripper.set(used)
self._samples.sample_in_gripper_name.set(name)
self._samples.sample_in_gripper_name.set(packed_name)
else:
getattr(self._samples.sample_placed, f"sample{slot}").set(used)
getattr(self._samples.sample_names, f"sample{slot}").set(name)
getattr(self._samples.sample_names, f"sample{slot}").set(packed_name)
# drop the cached state for this slot so the refresh() right after
# a write always repaints it, without waiting for the value to
# differ from a possibly-stale cache entry
@@ -374,17 +390,17 @@ class SampleStorageWidget(BECWidget, QWidget):
state = self._read_all_slots()
if not state:
return # bulk read failed; leave the current display untouched
for slot, (occupied, name) in state.items():
if self._last_state.get(slot) == (occupied, name):
for slot, (occupied, name, owner) in state.items():
if self._last_state.get(slot) == (occupied, name, owner):
continue
cell = self._cells.get(slot)
if cell is not None:
cell.set_state(occupied, name)
self._last_state[slot] = (occupied, name)
cell.set_state(occupied, name, owner)
self._last_state[slot] = (occupied, name, owner)
# ── mutation actions (called from _SlotCell context menu) ─────────────────
def action_change_name(self, slot: int, current_name: str) -> None:
def action_change_name(self, slot: int, current_name: str, current_owner: str = "") -> None:
"""Rename an occupied slot (stays occupied)."""
name, ok = QInputDialog.getText(
self,
@@ -403,7 +419,14 @@ class SampleStorageWidget(BECWidget, QWidget):
"Use “Clear (set empty)” to empty the slot instead.",
)
return
if self._write_slot(slot, 1, name):
owner, ok = QInputDialog.getText(
self,
"Sample owner",
f"Owner for slot {self._slot_title(slot)} (optional):",
text=current_owner,
)
owner = owner.strip() if ok else current_owner
if self._write_slot(slot, 1, name, owner):
self.refresh()
def action_new_sample(self, slot: int) -> None:
@@ -417,7 +440,11 @@ class SampleStorageWidget(BECWidget, QWidget):
if not name or name == EMPTY_NAME:
QMessageBox.warning(self, "Invalid name", "Please enter a non-empty sample name.")
return
if self._write_slot(slot, 1, name):
owner, ok = QInputDialog.getText(
self, "Sample owner", f"Owner for slot {self._slot_title(slot)} (optional):"
)
owner = owner.strip() if ok else ""
if self._write_slot(slot, 1, name, owner):
self.refresh()
def action_clear(self, slot: int) -> None:
csaxs_bec/bec_widgets/widgets/tomo_params/tomo_params.py
+18 -7
View File
@@ -1823,24 +1823,34 @@ class CommandJobBuilderDialog(QDialog):
def _requested_to_stepsize(angle_range: int, requested_total: int) -> float:
"""Convert a desired total projection count to tomo_angle_stepsize."""
"""Convert a desired total projection count to tomo_angle_stepsize.
angle_range is accepted for call-site compatibility but NOT used here:
N/step/total are always computed against a fixed 180 degrees,
independent of tomo_angle_range (tomo_angle_range only splits the 8
sub-tomograms into low/high halves in 360 mode -- it doesn't change
the total projection count or per-subtomo step). Mirrors
Flomni._subtomo_angle_plan()'s docstring / _tomo_type1_actual_grid().
"""
if requested_total <= 0:
return float(angle_range)
return (angle_range / requested_total) * 8
return 180.0
return (180.0 / requested_total) * 8
def _compute_type1(angle_range: int, stepsize: float) -> tuple[int, float, float]:
"""
Given stored tomo_angle_stepsize, return:
(actual_total, achievable_step, stepsize)
Mirrors the CLI's internal computation exactly.
Mirrors the CLI's internal computation exactly -- see
_requested_to_stepsize()'s docstring for why angle_range is accepted
but not used in the N/step/total math.
"""
if stepsize <= 0:
return 0, 0.0, 0.0
N = int(angle_range / stepsize)
N = int(180.0 / stepsize)
if N <= 0:
return 0, 0.0, 0.0
achievable_step = angle_range / N
achievable_step = 180.0 / N
actual_total = N * 8
return actual_total, achievable_step, stepsize
@@ -1849,7 +1859,8 @@ def _format_projections(params: dict) -> str:
"""
Human-readable projection count for a queued job, mirroring the CLI's
per-type logic:
- type 1: int(angle_range / stepsize) * 8 (8 equally spaced sub-tomos)
- type 1: int(180 / stepsize) * 8 (8 equally spaced sub-tomos,
independent of angle_range)
- type 3: int(180 / stepsize) * 8 (equally spaced, golden start)
- type 2: golden ratio has no fixed count -> configured max, or ∞ if unset
"""
csaxs_bec/devices/omny/flomni_sample_storage.py
+27 -8
View File
@@ -6,6 +6,8 @@ from ophyd import DynamicDeviceComponent as Dcpt
from ophyd import EpicsSignal
from prettytable import PrettyTable
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
class FlomniSampleStorageError(Exception):
pass
@@ -19,6 +21,9 @@ class FlomniSampleStorage(Device):
"unset_sample_slot",
"set_sample_in_gripper",
"unset_sample_in_gripper",
"get_sample_name",
"get_sample_owner",
"get_sample_name_and_owner",
"show_all",
]
SUB_VALUE = "value"
@@ -54,12 +59,12 @@ class FlomniSampleStorage(Device):
self.wait_for_connection()
self._run_subs(sub_type=self.SUB_VALUE, timestamp=timestamp, obj=self)
def set_sample_slot(self, slot_nr: int, name: str) -> bool:
def set_sample_slot(self, slot_nr: int, name: str, owner: str = "") -> bool:
if slot_nr > 20:
raise FlomniSampleStorageError(f"Invalid slot number {slot_nr}.")
getattr(self.sample_placed, f"sample{slot_nr}").set(1)
getattr(self.sample_names, f"sample{slot_nr}").set(name)
getattr(self.sample_names, f"sample{slot_nr}").set(pack_desc(name, owner))
def unset_sample_slot(self, slot_nr: int) -> bool:
if slot_nr > 20:
@@ -68,9 +73,9 @@ class FlomniSampleStorage(Device):
getattr(self.sample_placed, f"sample{slot_nr}").set(0)
getattr(self.sample_names, f"sample{slot_nr}").set("-")
def set_sample_in_gripper(self, name: str) -> bool:
def set_sample_in_gripper(self, name: str, owner: str = "") -> bool:
self.sample_in_gripper.set(1)
self.sample_in_gripper_name.set(name)
self.sample_in_gripper_name.set(pack_desc(name, owner))
def unset_sample_in_gripper(self) -> bool:
self.sample_in_gripper.set(0)
@@ -86,7 +91,15 @@ class FlomniSampleStorage(Device):
def get_sample_name(self, slot_nr) -> str:
val = getattr(self.sample_names, f"sample{slot_nr}").get()
return str(val)
return unpack_desc(str(val))[0]
def get_sample_owner(self, slot_nr) -> str:
val = getattr(self.sample_names, f"sample{slot_nr}").get()
return unpack_desc(str(val))[1]
def get_sample_name_and_owner(self, slot_nr) -> tuple:
val = getattr(self.sample_names, f"sample{slot_nr}").get()
return unpack_desc(str(val))
def show_all(self):
t = PrettyTable()
@@ -105,13 +118,19 @@ class FlomniSampleStorage(Device):
print("\n\nFollowing samples are currently loaded:\n")
for ct in range(1, 21):
if self.is_sample_slot_used(ct):
print(f" Position {ct:2.0f}: {self.get_sample_name(ct)}")
name, owner = self.get_sample_name_and_owner(ct)
owner_suffix = f" (owner: {owner})" if owner else ""
print(f" Position {ct:2.0f}: {name}{owner_suffix}")
if self.sample_in_gripper.get():
print(f"\n Gripper: {self.sample_in_gripper_name.get()}\n")
name, owner = unpack_desc(str(self.sample_in_gripper_name.get()))
owner_suffix = f" (owner: {owner})" if owner else ""
print(f"\n Gripper: {name}{owner_suffix}\n")
else:
print(f"\n Gripper: no sample\n")
if self.is_sample_slot_used(0):
print(f" flOMNI stage: {self.get_sample_name(0)}\n")
name, owner = self.get_sample_name_and_owner(0)
owner_suffix = f" (owner: {owner})" if owner else ""
print(f" flOMNI stage: {name}{owner_suffix}\n")
else:
print(f" flOMNI stage: no sample\n")
csaxs_bec/devices/omny/galil/fgalil_ophyd.py
+1
View File
@@ -41,6 +41,7 @@ class FlomniGalilController(GalilController):
"lights_off",
"lights_on",
"print_command_history",
"hard_abort_and_restore_positioning_mode",
]
def is_axis_moving(self, axis_Id, axis_Id_numeric) -> bool:
csaxs_bec/devices/omny/galil/galil_ophyd.py
+32
View File
@@ -46,6 +46,7 @@ class GalilController(Controller):
"is_thread_active",
"all_axes_referenced",
"print_command_history",
"hard_abort_and_restore_positioning_mode",
]
OKBLUE = "\033[94m"
@@ -100,6 +101,37 @@ class GalilController(Controller):
else:
return ":"
def hard_abort_and_restore_positioning_mode(
self, transfer_thread_id: int = 3, timeout: float = 5.0
) -> None:
"""
Hard abort of a running sample-transfer routine: stops all axes and
threads on the controller (XQ#STOP,1 via stop_all_axes(), which
halts the transfer thread, aborts all motion, and clears
mntprgs/mntmod), waits for the transfer thread to actually halt,
then switches the controller back to positioning mode (#POSMODE
refuses while mntprgs=1, so this must wait first).
Shared by both the CLI (Flomni.ftransfer_abort(), via Ctrl+C) and
the GUI hard-stop button, so there is exactly one implementation of
this safety-critical stop sequence -- two independent
implementations previously drifting apart is what let the GUI path
end up unsafe.
"""
self.stop_all_axes()
start = time.time()
while self.is_thread_active(transfer_thread_id):
if time.time() - start > timeout:
raise GalilError(
f"Hard abort requested but transfer thread {transfer_thread_id} did not "
f"stop within {timeout} s. Check the controller."
)
time.sleep(0.1)
self.socket_put_confirmed("XQ#POSMODE")
time.sleep(0.5)
if bool(float(self.socket_put_and_receive("MG mntmod").strip())):
raise GalilError("System is still in mount mode after hard abort.")
def get_digital_input(self, channel):
return bool(float(self.socket_put_and_receive(f"MG @IN[{channel}]").strip()))
csaxs_bec/devices/omny/sample_desc_codec.py
+52
View File
@@ -0,0 +1,52 @@
"""Pack/unpack a sample name + owner into a single EPICS DESC field.
Sample storage (flomni_sample_storage.py, omny_sample_storage.py) has only
one free-text field per slot -- the underlying record's DESC field, used as
the sample name. There is no separate PV for an owner, so an owner has to be
packed into that same string.
DESC_MAX_LEN is assumed (~40 chars is the generic EPICS base-record DESC
size), not confirmed against the real IOC .db (not present in this repo) --
verify against the real IOC before production rollout; the sim's mocked PVs
do not enforce any length limit.
"""
DESC_MAX_LEN = 40
EMPTY_SENTINEL = "-"
DELIMITER = " | "
def pack_desc(name: str, owner: str = "") -> str:
"""Pack (name, owner) into a single DESC string.
The empty-slot sentinel ("-") always passes through untouched, never
gets an owner appended. If the combined string doesn't fit
DESC_MAX_LEN, the name is truncated first (not the owner): owner is
typically a short, fixed-format identifier (e.g. an e-account) that
matters for accountability, while name is free text that already
tolerates truncation better.
"""
if name == EMPTY_SENTINEL:
return EMPTY_SENTINEL
if not owner:
return name[:DESC_MAX_LEN]
combined = f"{name}{DELIMITER}{owner}"
if len(combined) <= DESC_MAX_LEN:
return combined
name_budget = max(0, DESC_MAX_LEN - len(DELIMITER) - len(owner))
return f"{name[:name_budget]}{DELIMITER}{owner}"[:DESC_MAX_LEN]
def unpack_desc(raw: str) -> tuple[str, str]:
"""Unpack a raw DESC string into (name, owner).
Returns owner="" for a legacy string with no delimiter (no owner was
ever recorded), and passes the "-" empty-slot sentinel through
untouched with owner="".
"""
if raw is None or raw == EMPTY_SENTINEL:
return (EMPTY_SENTINEL, "")
if DELIMITER not in raw:
return (raw, "")
name, _, owner = raw.partition(DELIMITER)
return (name, owner)
omny_e2e_tests/sim_flomni_harness.py
+31
View File
@@ -417,6 +417,37 @@ def main(): # noqa: C901
"slot 1 -> gripper bookkeeping",
)
# --- 9b. sample owner (packed into the same DESC field as name) ------------------
samples.set_sample_slot(2, "owner_test_sample", owner="mholler")
check(
samples.get_sample_name(2) == "owner_test_sample", "get_sample_name returns plain name only"
)
check(samples.get_sample_owner(2) == "mholler", "get_sample_owner returns the packed owner")
check(
samples.get_sample_name_and_owner(2) == ("owner_test_sample", "mholler"),
"get_sample_name_and_owner returns (name, owner)",
)
samples.unset_sample_slot(2)
check(
str(getattr(samples.sample_names, "sample2").get()) == "-",
"unset_sample_slot writes bare '-', not '- | <owner>'",
)
# gripper move (slot -> gripper) must preserve owner, not just name. The
# gripper is a separate Cpt (sample_in_gripper_name), not part of the
# sample_names slot dict, so unpack it directly rather than via
# get_sample_name_and_owner (which only covers slots 0-20).
from csaxs_bec.devices.omny.sample_desc_codec import unpack_desc
samples.set_sample_slot(4, "transfer_test_sample", owner="e12345")
name, owner = samples.get_sample_name_and_owner(4)
samples.set_sample_in_gripper(name, owner=owner)
samples.unset_sample_slot(4)
check(
unpack_desc(str(samples.sample_in_gripper_name.get())) == ("transfer_test_sample", "e12345"),
"owner survives a slot -> gripper move",
)
samples.unset_sample_in_gripper()
frame_a = ids_cam.cam.get_image_data()
frame_b = ids_cam.cam.get_image_data()
check(not np.array_equal(frame_a, frame_b), "IDS frames are live (noise explicitly enabled)")
tests/tests_bec_ipython_client/test_flomni_tomo_angles.py
+136
View File
@@ -0,0 +1,136 @@
import numpy as np
import pytest
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni import Flomni
plan = Flomni._subtomo_angle_plan
STEPSIZES = [10.0, 7.0, 25.0, 12.5]
@pytest.mark.parametrize("stepsize", STEPSIZES)
def test_180_mode_regression(stepsize):
"""180-degree mode must be byte-for-byte identical to the original
scheme: all 8 sub-tomograms interlace the same [0,180) span at spacing
step/8, with no duplicates and no gaps."""
all_angles = []
for n in range(1, 9):
angles, offset, N, step = plan(n, 180, stepsize)
assert offset == 0
assert len(angles) == N
all_angles.append(angles)
combined = np.sort(np.concatenate(all_angles))
assert len(combined) == len(np.unique(np.round(combined, 6)))
diffs = np.diff(combined)
_, _, expected_total = _actual_grid(stepsize)
expected_step = 180.0 / (expected_total)
assert np.allclose(diffs, expected_step, atol=1e-6)
assert combined.min() >= 0
assert combined.max() < 180
@pytest.mark.parametrize("stepsize", STEPSIZES)
def test_360_mode_no_duplicates_and_no_internal_180_pairs(stepsize):
"""360-degree mode: no sub-tomogram may, by itself, contain two angles
exactly 180 degrees apart (that redundancy was the original bug), and
the combined 8-sub-tomogram set must contain no duplicate angles."""
all_angles = []
for n in range(1, 9):
angles, offset, N, step = plan(n, 360, stepsize)
assert offset == 0
assert len(angles) == N
assert angles.max() - angles.min() < 180
setc = set(np.round(angles, 6))
for a in setc:
assert round((a + 180) % 360, 6) not in setc
all_angles.append(angles)
combined = np.sort(np.concatenate(all_angles))
assert len(combined) == len(np.unique(np.round(combined, 6)))
@pytest.mark.parametrize("stepsize", STEPSIZES)
def test_no_angle_pair_360_degrees_apart_anywhere(stepsize):
"""No two of the 8 sub-tomograms' combined 360-degree angles may be
exactly 180 degrees apart from each other either -- a measurement
180 degrees from one already taken carries no new information, and
this must hold across the WHOLE combined set, not just within a
single sub-tomogram."""
combined = np.concatenate([plan(n, 360, stepsize)[0] for n in range(1, 9)])
setc = set(np.round(combined, 6))
for a in setc:
assert round((a + 180) % 360, 6) not in setc
@pytest.mark.parametrize("stepsize", STEPSIZES)
def test_total_projection_count_identical_180_vs_360(stepsize):
"""Total projections for a given tomo_angle_stepsize must be the same
whether tomo_angle_range is 180 or 360 (confirmed requirement)."""
_, _, n180 = _actual_grid(stepsize)
total_360 = sum(len(plan(n, 360, stepsize)[0]) for n in range(1, 9))
assert total_360 == n180
@pytest.mark.parametrize("stepsize", STEPSIZES)
def test_360_mode_mod_180_identical_to_180_mode(stepsize):
"""The core requirement: 360-degree mode's combined angle set, folded
mod 180, must be IDENTICAL (as a set, same count, no residue repeated)
to what 180-degree mode itself produces for the same stepsize -- zero
redundant measurements, full information content, same total count.
This requires the low half (subtomos 1,4,5,8) and high half (2,3,6,7)
to use COMPLEMENTARY phase positions, not the same ones shifted by
180 -- shared phases would make every low-half angle exactly 180
degrees from a high-half angle (the original bug)."""
angles180 = np.concatenate([plan(n, 180, stepsize)[0] for n in range(1, 9)])
angles360 = np.concatenate([plan(n, 360, stepsize)[0] for n in range(1, 9)])
folded = np.mod(np.round(angles360, 6), 180)
set180 = set(np.round(angles180, 6))
set_folded = set(np.round(folded, 6))
assert len(angles360) == len(angles180)
assert len(folded) == len(np.unique(folded)), "a mod-180 residue was hit more than once"
assert set_folded == set180
@pytest.mark.parametrize("stepsize", STEPSIZES)
@pytest.mark.parametrize("n", range(1, 9))
def test_360_mode_subtomo_keeps_same_phase_tier_as_180_mode(stepsize, n):
"""Subtomo N must carry the same phase_eighths tier in 360 mode as in
180 mode - e.g. subtomo 2 always represents the phase_eighths[2] tier,
whether that's within [0,180) (180 mode) or shifted +180 into
[180,360) (360 mode). A previous implementation reassigned phases by
acquisition-order-within-half instead of reusing the per-subtomo-number
value, so e.g. the angle physically corresponding to subtomo 2's tier
in 180 mode ended up under subtomo 4 in 360 mode - confusing, and
breaks any expectation that "subtomo N" means the same thing across
modes."""
angles180, _, N, step = plan(n, 180, stepsize)
angles360, _, _, _ = plan(n, 360, stepsize)
base = 0.0 if n % 4 in (1, 0) else 180.0
offset180 = set(np.round(np.mod(angles180, step), 6))
offset360 = set(np.round(np.mod(angles360 - base, step), 6))
assert offset360 == offset180
@pytest.mark.parametrize("tomo_angle_range", [180, 360])
@pytest.mark.parametrize("stepsize", STEPSIZES)
def test_resume_round_trip(tomo_angle_range, stepsize):
"""Resuming with an explicit start_angle must recover the exact loop
index that originally produced that angle, for every sub-tomogram and
every position -- including the float tie-break cases (subtomo 2 in
180 mode; subtomos 3 and 6 in 360 mode, both landing on phase/step ==
0.5)."""
for n in range(1, 9):
angles, _, N, _ = plan(n, tomo_angle_range, stepsize)
for i in range(N):
_, offset, _, _ = plan(n, tomo_angle_range, stepsize, start_angle=angles[i])
assert offset == i, f"n={n} i={i} range={tomo_angle_range} stepsize={stepsize}"
def _actual_grid(stepsize):
N = int(180.0 / stepsize)
step = 180.0 / N
return N, step, N * 8
tests/tests_bec_ipython_client/test_sample_desc_codec.py
+59
View File
@@ -0,0 +1,59 @@
import pytest
from csaxs_bec.devices.omny.sample_desc_codec import DESC_MAX_LEN, pack_desc, unpack_desc
def test_round_trip():
packed = pack_desc("my_sample", "mholler")
assert packed == "my_sample | mholler"
assert unpack_desc(packed) == ("my_sample", "mholler")
def test_no_owner_no_trailing_delimiter():
"""pack_desc(name, "") must equal name exactly -- required so the
raw-copy call sites (e.g. gripper<->slot transfer) stay no-ops when no
owner has ever been set."""
assert pack_desc("my_sample") == "my_sample"
assert pack_desc("my_sample", "") == "my_sample"
def test_legacy_string_no_delimiter():
"""A pre-existing sample name with no delimiter unpacks to (name, "")."""
assert unpack_desc("my_sample") == ("my_sample", "")
def test_empty_sentinel_round_trip():
assert pack_desc("-") == "-"
assert pack_desc("-", "someone") == "-" # never pack an owner onto the sentinel
assert unpack_desc("-") == ("-", "")
assert unpack_desc(None) == ("-", "")
@pytest.mark.parametrize("owner", ["", "mholler"])
def test_truncation_long_name(owner):
long_name = "x" * 60
packed = pack_desc(long_name, owner)
assert len(packed) <= DESC_MAX_LEN
name, unpacked_owner = unpack_desc(packed)
assert unpacked_owner == owner
assert name == long_name[: len(name)] # truncated prefix of the original name
def test_truncation_long_owner():
packed = pack_desc("s", "y" * 60)
assert len(packed) <= DESC_MAX_LEN
def test_truncation_no_owner_no_wasted_delimiter():
"""Truncating a too-long name with no owner must not waste characters
on a trailing delimiter that has nothing after it."""
packed = pack_desc("x" * 60, "")
assert packed == "x" * DESC_MAX_LEN
assert not packed.endswith(" | ")
def test_pack_stays_within_max_len_for_typical_values():
for name_len in range(0, 45, 5):
for owner_len in range(0, 20, 5):
packed = pack_desc("n" * name_len, "o" * owner_len if owner_len else "")
assert len(packed) <= DESC_MAX_LEN
tests/tests_bec_ipython_client/test_tomo_params_widget_math.py
+48
View File
@@ -0,0 +1,48 @@
"""Regression test for TomoParamsWidget's independent (duplicated)
projection-count math: it must stay in sync with Flomni's own
_tomo_type1_actual_grid()/_subtomo_angle_plan() -- this exact drift (the
widget kept using the old, angle-range-dependent formula after flomni.py
was fixed to be mode-independent) is what caused the GUI to show the
projection count doubling when switching from 180 to 360 mode.
"""
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_type1,
_requested_to_stepsize,
)
STEPSIZES = [10.0, 7.0, 25.0, 12.5]
@pytest.mark.parametrize("stepsize", STEPSIZES)
@pytest.mark.parametrize("angle_range", [180, 360])
def test_compute_type1_matches_flomni(stepsize, angle_range):
total, step, N = _flomni_reference(stepsize)
actual_total, achievable_step, _ = _compute_type1(angle_range, stepsize)
assert actual_total == total
assert achievable_step == pytest.approx(step)
@pytest.mark.parametrize("requested_total", [24, 48, 96, 144, 200])
@pytest.mark.parametrize("angle_range", [180, 360])
def test_requested_to_stepsize_matches_flomni_wizard_formula(requested_total, angle_range):
"""Mirrors the exact line in Flomni.tomo_parameters()'s wizard:
self.tomo_angle_stepsize = (180.0 / tomo_numberofprojections) * 8"""
stepsize = _requested_to_stepsize(angle_range, requested_total)
assert stepsize == pytest.approx((180.0 / requested_total) * 8)
def test_total_projections_identical_between_modes_for_same_stepsize():
for stepsize in STEPSIZES:
total_180, _, _ = _compute_type1(180, stepsize)
total_360, _, _ = _compute_type1(360, stepsize)
assert total_180 == total_360
def _flomni_reference(stepsize):
N = int(180.0 / stepsize)
step = 180.0 / N
return N * 8, step, N