fix(flomni): stop duplicate 360deg tomo angles, add sample owner, fix abort crash
Three independent fixes:
- sub_tomo_scan(): in 360-degree tomo_type-1 mode, every sub-tomogram used
to sweep the full 0-360 range, so each one's own fine grid contained
angle pairs exactly 180deg apart -- redundant tomographic information.
Sub-tomograms now each cover a 180deg span, split into low/high halves
by subtomo_number % 4 with a bit-reversal-of-4 phase table, so adjacent
pairs (1,2)/(3,4)/(5,6)/(7,8), either quartet, and all 8 combined each
independently form a complete, evenly-spaced 360deg tomogram at
successively finer spacing. Total projection count for a given
tomo_angle_stepsize is now identical between 180 and 360 mode (same N,
no longer doubled). Updated the 4 other consumers of the old N/step
formula (zero-deg reference gating, _tomo_type1_actual_grid, the
parameter wizard, the PDF report) to match. 180-degree mode is
unchanged. Live-verified against the running flomni sim: real motor
motion traces the expected boustrophedon path with no duplicate or
180deg-apart angles.
- Sample storage: added an owner field, packed into the same EPICS DESC
field as the sample name ("name | owner", via new sample_desc_codec.py)
since there's no separate PV for it. Wired through
FlomniSampleStorage, the CLI (flomni_modify_storage_non_interactive,
ftransfer_modify_storage), the two transfer routines that forward a
raw DESC value across a gripper move (now unpacked/repacked so owner
survives the move instead of being dropped or double-packed), and
SampleStorageWidget. Scoped to flomni only this session; OMNY's
storage/transfer mixin is unchanged.
- ConsoleButtonsWidget's ABORT button used to send SIGINT then, 500ms
later, a stop_devices() broadcast to ALL devices with no stop_id --
an un-suppressed error from that broadcast landing on a queue-tracked
instruction could kill the scan worker thread outright, requiring a
full BEC restart. Replaced with: queue.request_scan_abortion() (safe
no-op if idle, but registers a stop_id so expected errors are
suppressed), then SIGINT, then a direct, immediate Galil hard stop via
new GalilController.hard_abort_and_restore_positioning_mode() -- the
same method ftransfer_abort() now delegates to, so the CLI and GUI
paths can't drift apart again. SIGINT is sent before the hard stop:
live testing showed that if the hard stop's mntprgs-clearing side
effect lands first, a same-session polling loop can mistake it for
normal completion and fall through into ensure_gripper_up(), which
must not happen mid-transfer -- sending SIGINT first (near-instant)
gives that loop's own KeyboardInterrupt handler a head start before
the hard stop's own multi-step sequence completes. The button is
labeled per beamline (e.g. "Flomni Motion Stop") and disabled rather
than silently inert when no hard-stop device is configured/enabled.
Scoped to flomni only this session (OMNY/LamNI wiring deferred).
Live-verified against the running sim, including the exact
stop-lands-mid-queued-instruction scenario that previously crashed
the scan worker.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
x01dc
@@ -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,121 @@ 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)), with the 4 sub-tomograms
|
||||
serving each half interlaced via a bit-reversal of {0,1,2,3}
|
||||
(not sequential order): this specific permutation is what makes
|
||||
adjacent pairs (1,2)/(3,4)/(5,6)/(7,8), either quartet (1-4 or
|
||||
5-8), and all 8 combined each independently form a complete,
|
||||
evenly-spaced 360-degree tomogram at successively finer spacing
|
||||
(step, step/2, step/4) - a sequential assignment would leave gaps
|
||||
in the quartet-level grids. Without this split, every
|
||||
sub-tomogram's own 360-degree sweep would contain angle pairs
|
||||
exactly 180 degrees apart, which is redundant tomographic
|
||||
information.
|
||||
|
||||
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
|
||||
|
||||
if tomo_angle_range == 180:
|
||||
base = 0.0
|
||||
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]
|
||||
forward = bool(subtomo_number % 2)
|
||||
else:
|
||||
quarter = {1: 0, 4: 2, 5: 1, 8: 3, 2: 0, 3: 2, 6: 1, 7: 3}
|
||||
mod4 = subtomo_number % 4
|
||||
base = 0.0 if mod4 in (1, 0) else 180.0
|
||||
phase = step / 4.0 * quarter[subtomo_number]
|
||||
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 +2424,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 +2454,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.
|
||||
@@ -2632,14 +2657,23 @@ class Flomni(
|
||||
# 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:
|
||||
if (
|
||||
start_angle is None
|
||||
and self._subtomo_starts_near_zero(ii)
|
||||
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
|
||||
# every time the rotation passes back through 0, 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.
|
||||
# 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
|
||||
@@ -3537,11 +3571,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 +3593,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):
|
||||
@@ -3695,7 +3734,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
|
||||
@@ -4337,15 +4379,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",
|
||||
|
||||
@@ -131,9 +131,18 @@ 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.
|
||||
self.console = self.gui.flomni.new(
|
||||
"ConsoleButtonsWidget", object_name="console", where="bottom"
|
||||
"ConsoleButtonsWidget",
|
||||
object_name="console",
|
||||
where="bottom",
|
||||
hard_stop_device_name="ftransy",
|
||||
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
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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,50 @@ 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.
|
||||
|
||||
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 configured, or if that device isn't present/enabled in the
|
||||
current session, the button is disabled rather than silently doing
|
||||
nothing.
|
||||
"""
|
||||
|
||||
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._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 +68,25 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
|
||||
self._client_pid = os.getppid()
|
||||
self._init_ui()
|
||||
|
||||
def _hard_stop_available(self) -> bool:
|
||||
"""True if hard_stop_device_name names a device that's actually
|
||||
present and enabled in this session -- checked once at construction
|
||||
time (the config doesn't change mid-session).
|
||||
|
||||
Device access is wrapped in try/except: DeviceManagerBase.__getattr__
|
||||
raises DeviceConfigError (not AttributeError) for an unknown device
|
||||
name, so a plain getattr(self.dev, name, None) would NOT fall back
|
||||
to the default and would propagate instead (same guard as
|
||||
SampleStorageWidget._check_flomni_available()).
|
||||
"""
|
||||
if not self._hard_stop_device_name:
|
||||
return False
|
||||
try:
|
||||
device = getattr(self.dev, self._hard_stop_device_name, None)
|
||||
except Exception:
|
||||
return False
|
||||
return device is not None and getattr(device, "enabled", True)
|
||||
|
||||
def _init_ui(self):
|
||||
layout = QVBoxLayout(self)
|
||||
|
||||
@@ -81,11 +111,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 +153,42 @@ 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 -- not delayed by anything above: sending
|
||||
# SIGINT is a near-instant os.kill() call, so this still runs
|
||||
# effectively immediately.
|
||||
if self._hard_stop_device_name:
|
||||
try:
|
||||
device = getattr(self.dev, self._hard_stop_device_name, None)
|
||||
except Exception:
|
||||
device = None
|
||||
if device is not None:
|
||||
logger.warning(
|
||||
f"ConsoleButtonsWidget: hard-stopping {self._hard_stop_device_name}"
|
||||
)
|
||||
try:
|
||||
device.controller.hard_abort_and_restore_positioning_mode()
|
||||
except Exception:
|
||||
logger.exception("ConsoleButtonsWidget: hard motion stop failed")
|
||||
|
||||
@SafeProperty(str)
|
||||
def message(self):
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()))
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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)")
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
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 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_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_full_grid_evenly_spaced(stepsize):
|
||||
"""All 8 sub-tomograms combined must be a complete, evenly-spaced grid
|
||||
covering the full [0,360) range at spacing step/4, no gaps."""
|
||||
_, step180, _ = _actual_grid(stepsize)
|
||||
all_angles = np.sort(np.concatenate([plan(n, 360, stepsize)[0] for n in range(1, 9)]))
|
||||
diffs = np.diff(all_angles)
|
||||
assert np.allclose(diffs, step180 / 4.0, atol=1e-6)
|
||||
assert all_angles.min() >= 0
|
||||
assert all_angles.max() < 360
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stepsize", STEPSIZES)
|
||||
@pytest.mark.parametrize("pair", [(1, 2), (3, 4), (5, 6), (7, 8)])
|
||||
def test_360_mode_adjacent_pair_is_complete_coarse_tomogram(stepsize, pair):
|
||||
"""Any adjacent pair (1,2)/(3,4)/(5,6)/(7,8) must independently
|
||||
reconstruct a complete, evenly-spaced 360-degree tomogram at the
|
||||
coarse (unrefined) step spacing."""
|
||||
_, step180, _ = _actual_grid(stepsize)
|
||||
a, b = pair
|
||||
combined = np.sort(np.concatenate([plan(a, 360, stepsize)[0], plan(b, 360, stepsize)[0]]))
|
||||
diffs = np.diff(combined)
|
||||
assert np.allclose(diffs, step180, atol=1e-6)
|
||||
assert combined.min() >= 0
|
||||
assert combined.max() < 360
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stepsize", STEPSIZES)
|
||||
@pytest.mark.parametrize("quartet", [(1, 2, 3, 4), (5, 6, 7, 8)])
|
||||
def test_360_mode_quartet_is_complete_half_sampled_tomogram(stepsize, quartet):
|
||||
"""Either quartet (1-4 or 5-8) must independently reconstruct a
|
||||
complete, evenly-spaced 360-degree tomogram at half the finest
|
||||
(8-subtomo) spacing."""
|
||||
_, step180, _ = _actual_grid(stepsize)
|
||||
combined = np.sort(np.concatenate([plan(n, 360, stepsize)[0] for n in quartet]))
|
||||
diffs = np.diff(combined)
|
||||
assert np.allclose(diffs, step180 / 2.0, atol=1e-6)
|
||||
assert combined.min() >= 0
|
||||
assert combined.max() < 360
|
||||
|
||||
|
||||
@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 4 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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user