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

feat(flomni): tomo scan queue, crash-safe resume, idle-time-aware ETA, uniform angular step fix

Browse Source 
- Tomo scan queue: queue multiple parameter sets and run them sequentially,
  unattended, on the same sample (tomo_queue_add/show/execute/delete/clear);
  resumes a partially-completed job automatically rather than restarting it
- tomo_scan_resume(): resume a crashed/interrupted tomo scan from the exact
  subtomo/angle it stopped at, instead of restarting from the beginning
- Idle-time-aware ETA: detects gaps (crashes, beamline-down pauses) and
  excludes them from the remaining-time estimate; reports total time lost
  to gaps at the end of a scan
- Fixed non-uniform angular spacing in the interlaced 8-sub-tomogram
  tomogram (sub_tomo_scan): phase offsets and per-projection step now
  derive from the same corrected value, independent of requested total
- Fixed corr_pos_y/corr_angle_y/corr_pos_y_2/corr_angle_y_2/
  tomo_alignment_fit being silently wiped on every client restart
  (XrayEyeAlign.__init__); tomo_alignment_fit now also resets at sample
  change, where it was previously missed
- Tracked, stepped fsamy moves (umvr_fsamy_tracked/umv_fsamy_tracked) to
  keep the laser tracker locked during large moves, used in sample
  transfer and manual x-ray-eye alignment jogs
- Added zero_deg_reference_at_each_subtomo for radiation-damage tracking
- tomo_parameters(): fixed several display/rounding inconsistencies and
  added a notification when the requested projection count gets adjusted
This commit is contained in:
x12sa
2026-06-30 14:31:36 +02:00
committed by holler
co-authored by holler
parent 4e472a9a6a
commit b362842e4f
1 changed files with 608 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+608 -101
View File
@@ -54,34 +54,6 @@ class FlomniError(Exception):
pass
# class FlomniTools:
# def yesno(self, message: str, default="none", autoconfirm=0) -> bool:
# if autoconfirm and default == "y":
# self.printgreen(message + " Automatically confirming default: yes")
# return True
# elif autoconfirm and default == "n":
# self.printgreen(message + " Automatically confirming default: no")
# return False
# if default == "y":
# message_ending = " [Y]/n? "
# elif default == "n":
# message_ending = " y/[N]? "
# else:
# message_ending = " y/n? "
# while True:
# user_input = input(self.OKBLUE + message + message_ending + self.ENDC)
# if (
# user_input == "Y" or user_input == "y" or user_input == "yes" or user_input == "Yes"
# ) or (default == "y" and user_input == ""):
# return True
# if (
# user_input == "N" or user_input == "n" or user_input == "no" or user_input == "No"
# ) or (default == "n" and user_input == ""):
# return False
# else:
# print("Please expicitely confirm y or n.")
class FlomniInitStagesMixin:
def flomni_init_stages(self):
@@ -380,7 +352,7 @@ class FlomniInitStagesMixin:
umv(dev.fsamy, flomni_samy_in)
# after init reduce vertical stage speed
dev.fsamy.controller.socket_put_confirmed("axspeed[5]=5000")
dev.fsamy.controller.socket_put_confirmed("axspeed[5]=20000")
umv(dev.feyey, -8)
@@ -452,6 +424,7 @@ class FlomniSampleTransferMixin:
if not sample_in_position:
raise FlomniError("There is no sample in the sample stage. Aborting.")
self.reset_correction()
self.reset_tomo_alignment_fit()
dev.rtx.controller.feedback_disable()
self.ensure_fheater_up()
self.ensure_gripper_up()
@@ -488,6 +461,56 @@ class FlomniSampleTransferMixin:
def laser_tracker_off(self):
dev.rtx.controller.laser_tracker_off()
def umvr_fsamy_tracked(self, shift: float, step: float = 0.01):
"""
Relative move of fsamy by `shift` mm, broken into steps of at most
`step` mm (default 0.01 mm = 10 um), calling laser_tracker_on()
after every step so the tracker stays locked during the move
(fsamy travels a lot during sample change and the tracker can't
keep up in one big jump). fsamy itself moves in mm, so shift/step
are taken directly in mm -- no unit conversion needed.
Uses dev.rtx.controller.laser_tracker_on() directly rather than
self.laser_tracker_on() (which adds a 0.2s sleep plus a
signal-strength check) since this runs once per small step and that
overhead would add up across what can be 100+ steps for a typical
sample-change-sized move; the goal here is just to keep the tracker
re-acquiring as fsamy moves, not to verify signal strength at every
single step.
Args:
shift: relative move distance in mm (signed; negative moves the
other direction).
step: maximum step size in mm per tracker re-acquisition.
Must be positive. Default 0.01 mm (10 um).
"""
if step <= 0:
raise ValueError("step must be a positive number of mm.")
direction = 1 if shift >= 0 else -1
remaining = abs(shift)
while remaining > 0:
this_step = min(step, remaining)
scans.umv(dev.fsamy, direction * this_step, relative=True)
dev.rtx.controller.laser_tracker_on()
remaining -= this_step
def umv_fsamy_tracked(self, target: float, step: float = 0.01):
"""
Absolute move of fsamy to `target` mm (same units as dev.fsamy's
readback / dev.fsamy.user_parameter.get("in")), broken into steps of
at most `step` mm via umvr_fsamy_tracked().
Args:
target: absolute target position in mm.
step: maximum step size in mm per tracker re-acquisition.
Must be positive (validated in umvr_fsamy_tracked).
Default 0.01 mm (10 um).
"""
current = dev.fsamy.readback.get()
shift_mm = target - current
self.umvr_fsamy_tracked(shift_mm, step)
def show_signal_strength_interferometer(self):
dev.rtx.controller.show_signal_strength_interferometer()
@@ -534,16 +557,13 @@ class FlomniSampleTransferMixin:
self.check_tray_in()
self.laser_tracker_off()
time.sleep(0.05)
fsamy_in = dev.fsamy.user_parameter.get("in")
if fsamy_in is None:
raise FlomniError(
"Could not find an 'IN' position for fsamy. Please check your config."
)
umv(dev.fsamy, fsamy_in)
time.sleep(0.05)
self.laser_tracker_on()
self.umv_fsamy_tracked(fsamy_in)
time.sleep(0.05)
self.laser_tracker_off()
time.sleep(0.05)
@@ -618,7 +638,7 @@ class FlomniSampleTransferMixin:
self.check_sensor_connected()
sample_in_gripper = dev.flomni_samples.is_sample_in_gripper()
# bool(float(dev.flomni_samples.sample_in_gripper.get()))
# dev.flomni_samples.sample_in_gripper.get()
if not sample_in_gripper:
raise FlomniError("The gripper does not carry a sample.")
@@ -991,6 +1011,46 @@ class FlomniAlignmentMixin:
/ default_correction_file_rel
).resolve()
# --- additional y-correction curves: global-var-backed so they survive
# a BEC client restart, same pattern as every other persistent setting
# in this class. NOTE: must NOT be pre-assigned in Flomni.__init__ -
# doing so would overwrite the persisted value with [] on every restart.
@property
def corr_pos_y(self):
val = self.client.get_global_var("corr_pos_y")
return [] if val is None else val
@corr_pos_y.setter
def corr_pos_y(self, val: list):
self.client.set_global_var("corr_pos_y", val)
@property
def corr_angle_y(self):
val = self.client.get_global_var("corr_angle_y")
return [] if val is None else val
@corr_angle_y.setter
def corr_angle_y(self, val: list):
self.client.set_global_var("corr_angle_y", val)
@property
def corr_pos_y_2(self):
val = self.client.get_global_var("corr_pos_y_2")
return [] if val is None else val
@corr_pos_y_2.setter
def corr_pos_y_2(self, val: list):
self.client.set_global_var("corr_pos_y_2", val)
@property
def corr_angle_y_2(self):
val = self.client.get_global_var("corr_angle_y_2")
return [] if val is None else val
@corr_angle_y_2.setter
def corr_angle_y_2(self, val: list):
self.client.set_global_var("corr_angle_y_2", val)
def reset_correction(self, use_default_correction=True):
"""
Reset the correction to the default values.
@@ -1238,6 +1298,7 @@ class _ProgressProxy:
"estimated_remaining_time": None,
"estimated_finish_time": None,
"heartbeat": None,
"accumulated_idle_time": 0.0,
}
def __init__(self, client):
@@ -1287,6 +1348,63 @@ class _ProgressProxy:
return self._load()
class _TomoQueueProxy:
"""List-like proxy that persists a queue of tomo parameter snapshots as a
BEC global variable, mirroring the pattern used by :class:`_ProgressProxy`.
Each entry is a plain dict: ``{"label": ..., "params": {...}, "status":
..., "added_at": ...}``, where ``params`` holds a snapshot of all
global-var-backed tomo scan parameters (see
``Flomni._TOMO_QUEUE_PARAM_NAMES``). Stored as a single list under
``tomo_queue`` so the queue is visible from any BEC client session via
``client.get_global_var("tomo_queue")`` and survives a kernel restart.
"""
_GLOBAL_VAR_KEY = "tomo_queue"
def __init__(self, client):
self._client = client
def _load(self) -> list:
val = self._client.get_global_var(self._GLOBAL_VAR_KEY)
if val is None:
return []
return val
def _save(self, jobs: list) -> None:
self._client.set_global_var(self._GLOBAL_VAR_KEY, jobs)
def as_list(self) -> list:
"""Return a plain copy of the current queue."""
return self._load()
def append(self, job: dict) -> int:
jobs = self._load()
jobs.append(job)
self._save(jobs)
return len(jobs) - 1
def pop(self, index: int) -> dict:
jobs = self._load()
job = jobs.pop(index)
self._save(jobs)
return job
def update(self, index: int, **kwargs) -> None:
jobs = self._load()
jobs[index].update(kwargs)
self._save(jobs)
def clear(self) -> None:
self._save([])
def __len__(self) -> int:
return len(self._load())
def __getitem__(self, index):
return self._load()[index]
class Flomni(
FlomniInitStagesMixin,
FlomniSampleTransferMixin,
@@ -1305,12 +1423,17 @@ class Flomni(
self.special_angle_tolerance = 20
self._current_special_angles = []
self._beam_is_okay = True
self.corr_pos_y = []
self.corr_angle_y = []
self.corr_pos_y_2 = []
self.corr_angle_y_2 = []
self._progress_proxy = _ProgressProxy(self.client)
self._progress_proxy.reset()
# Deliberately NOT calling reset() here: this dict is persisted via a
# BEC global var specifically so it survives a kernel restart (e.g.
# after a crash), which is what tomo_scan_resume() relies on.
# Resetting it unconditionally on every Flomni() instantiation wiped
# tomo_start_time (and everything else) on exactly the restarts where
# resuming matters most. A genuinely new tomo_scan() call already
# resets the relevant fields itself (see the "new scan" branch
# below); use tomo_progress_reset() if you want to explicitly clear
# stale progress without starting a new scan.
self._tomo_queue_proxy = _TomoQueueProxy(self.client)
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
FlomniWebpageGenerator,
)
@@ -1606,6 +1729,27 @@ class Flomni(
def golden_projections_at_0_deg_for_damage_estimation(self, val: float):
self.client.set_global_var("golden_projections_at_0_deg_for_damage_estimation", val)
@property
def zero_deg_reference_at_each_subtomo(self):
"""If True (tomo_type == 1 only), an additional projection at exactly
0 degrees is acquired at the start of every odd (forward)
sub-tomogram - i.e. every time the rotation passes back through
0 degrees - and once more after the final (8th) sub-tomogram
completes. Together with sub-tomogram 1's own first projection
(which already lands exactly on 0 degrees), this gives a 0-degree
reference shot at every natural pass through 0 across the full
tomogram, useful for tracking radiation damage over time. Mirrors
golden_projections_at_0_deg_for_damage_estimation, which provides
similar functionality for tomo_type 2/3."""
val = self.client.get_global_var("zero_deg_reference_at_each_subtomo")
if val is None:
return False
return val
@zero_deg_reference_at_each_subtomo.setter
def zero_deg_reference_at_each_subtomo(self, val: bool):
self.client.set_global_var("zero_deg_reference_at_each_subtomo", val)
@property
def golden_ratio_bunch_size(self):
val = self.client.get_global_var("golden_ratio_bunch_size")
@@ -1700,7 +1844,7 @@ class Flomni(
umv(dev.fsamroy, 0)
self.OMNYTools.printgreenbold(
"\n\nAlignment scan finished. Please run SPEC_ptycho_align and load the new fit by flomni.read_alignment_offset() ."
"\n\nAlignment scan finished. Please run BEC_ptycho_align and load the new fit by flomni.read_alignment_offset() ."
)
# summary of alignment scan numbers
@@ -1755,37 +1899,61 @@ class Flomni(
if explicit_start_angle:
print(f"Sub tomo scan with start angle {start_angle} requested.")
max_allowed_angle = self.tomo_angle_range + 0.05 + self.tomo_angle_stepsize
# 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 = self.tomo_angle_stepsize / 8.0 * 4
start_angle = step / 8.0 * 4
elif subtomo_number == 3:
start_angle = self.tomo_angle_stepsize / 8.0 * 2
start_angle = step / 8.0 * 2
elif subtomo_number == 4:
start_angle = self.tomo_angle_stepsize / 8.0 * 6
start_angle = step / 8.0 * 6
elif subtomo_number == 5:
start_angle = self.tomo_angle_stepsize / 8.0 * 1
start_angle = step / 8.0 * 1
elif subtomo_number == 6:
start_angle = self.tomo_angle_stepsize / 8.0 * 5
start_angle = step / 8.0 * 5
elif subtomo_number == 7:
start_angle = self.tomo_angle_stepsize / 8.0 * 3
start_angle = step / 8.0 * 3
elif subtomo_number == 8:
start_angle = self.tomo_angle_stepsize / 8.0 * 7
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, so shift the
# freshly-computed phase up by one full angular range. 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 = min(start_angle + self.tomo_angle_range, max_allowed_angle)
# 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
@@ -1793,19 +1961,19 @@ class Flomni(
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
angle_end = min(start + self.tomo_angle_range, max_allowed_angle)
span = angle_end - start
angles = start + np.arange(N) * step
else: # even = reverse: high -> low
min_allowed_angle = 0
angle_end = max(start - self.tomo_angle_range, min_allowed_angle)
span = start - angle_end
# number of projections needed to maintain step size
N = int(span / self.tomo_angle_stepsize) + 1
angles = np.linspace(start, angle_end, num=N, endpoint=True)
angles = start - np.arange(N) * step
for i, angle in enumerate(angles):
@@ -1813,32 +1981,41 @@ class Flomni(
# --- NEW LOGIC FOR OFFSET WHEN start_angle IS SPECIFIED ---
if i == 0:
step = self.tomo_angle_stepsize
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 / step)
self._subtomo_offset = round((start_angle - phase) / step)
else: # even = reverse direction
self._subtomo_offset = round((self.tomo_angle_range - start_angle) / step)
self._subtomo_offset = round(
((phase + self.tomo_angle_range - step) - start_angle) / step
)
# progress index must always increase
self.progress["subtomo_projection"] = self._subtomo_offset + i
# ------------------------------------------------------------
# existing progress fields
self.progress["subtomo_total_projections"] = int(
self.tomo_angle_range / self.tomo_angle_stepsize
)
# existing progress fields. N is already an int (by
# construction, see above), so total_projections = N * 8 is
# exact - no float round-trip noise, and no separate round()/
# int() cast needed here.
self.progress["subtomo_total_projections"] = N
self.progress["projection"] = (subtomo_number - 1) * self.progress[
"subtomo_total_projections"
] + self.progress["subtomo_projection"]
self.progress["total_projections"] = (
self.tomo_angle_range / self.tomo_angle_stepsize
) * 8
self.progress["total_projections"] = N * 8
self.progress["angle"] = angle
# finally do the scan at this angle
@@ -1848,7 +2025,29 @@ class Flomni(
def _tomo_scan_at_angle(self, angle, subtomo_number):
if 0 <= angle < self.tomo_angle_range + 0.05:
self.progress["heartbeat"] = datetime.datetime.now().isoformat()
now = datetime.datetime.now()
prev_heartbeat_str = self.progress.get("heartbeat")
if prev_heartbeat_str is not None:
gap = (now - datetime.datetime.fromisoformat(prev_heartbeat_str)).total_seconds()
# Normal cadence between consecutive projections is roughly
# the acquisition time plus motor/readout overhead. A gap
# well beyond that means something interrupted the scan in
# between (beamline-down interlock pause, a crash + manual
# restart, ...) -- attribute the excess to idle time so it
# doesn't drag down the apparent scan rate used for the ETA
# below. The 5x/60s margins are a heuristic, not a precise
# timing model -- tune if it over/under-triggers in practice.
normal_cadence = max(60.0, 5 * self.tomo_countingtime * self.frames_per_trigger)
if gap > normal_cadence:
idle = gap - normal_cadence
self.progress["accumulated_idle_time"] = (
self.progress.get("accumulated_idle_time", 0.0) + idle
)
print(
f"Detected a {self._format_duration(gap)} gap since the last projection"
f" -- excluding {self._format_duration(idle)} from the ETA estimate."
)
self.progress["heartbeat"] = now.isoformat()
print(f"Starting flOMNI scan for angle {angle} in subtomo {subtomo_number}")
self._print_progress()
@@ -1919,15 +2118,34 @@ class Flomni(
# accumulated enough projections to compute a fresh one
self.progress["estimated_remaining_time"] = None
self.progress["estimated_finish_time"] = None
self.progress["accumulated_idle_time"] = 0.0
self.progress["heartbeat"] = None
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
@@ -2025,6 +2243,58 @@ class Flomni(
self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
self._print_progress()
self.OMNYTools.printgreenbold("Tomoscan finished")
print(
f"Total measurement time lost to detected gaps: {self._format_duration(self.progress.get('accumulated_idle_time', 0.0))}"
)
def tomo_scan_resume(self) -> None:
"""Resume a tomo_scan() that crashed or was interrupted, picking up
automatically from wherever ``progress`` last reported -- no need to
read the last subtomo/angle/projection off the progress printout by
hand and pass it to tomo_scan() yourself.
Re-attempts the exact angle/projection that was in progress at the
moment of interruption, rather than skipping ahead to the next one:
from here we can't be sure whether that last point was actually
acquired before the crash, and re-acquiring one extra projection is
harmless.
Works standalone for a normal tomo_scan(), independent of the tomo
queue. tomo_queue_execute() also calls this internally when resuming
a job that previously failed partway through, so a failed queue job
picks up mid-scan rather than restarting from subtomo 1 / angle 0.
"""
if self.progress.get("tomo_start_time") is None:
print("No tomo scan in progress to resume -- nothing to do.")
return
if self.tomo_type == 1:
subtomo_start = self.progress["subtomo"]
start_angle = self.progress["angle"]
if subtomo_start < 1:
print("No tomo scan in progress to resume -- nothing to do.")
return
print(f"Resuming tomo scan at subtomo {subtomo_start}, angle {start_angle:.3f} deg.")
self.tomo_scan(subtomo_start=subtomo_start, start_angle=start_angle)
elif self.tomo_type in (2, 3):
projection_number = self.progress["projection"]
print(f"Resuming tomo scan at projection {projection_number}.")
self.tomo_scan(projection_number=projection_number)
else:
raise FlomniError("undefined tomo type")
def tomo_progress_reset(self) -> None:
"""Explicitly clear the persisted tomo progress (start time, ETA,
current angle/subtomo/projection, accumulated idle time, ...).
Not called automatically anymore on Flomni() startup -- that used to
wipe an in-progress scan's state on every kernel restart, which is
exactly the state tomo_scan_resume() needs. Call this by hand if you
want a clean progress display without it being tied to starting a
new tomo_scan() (which already resets the relevant fields itself).
"""
self._progress_proxy.reset()
print("Tomo progress reset.")
@staticmethod
def _format_duration(seconds: float) -> str:
@@ -2046,6 +2316,11 @@ class Flomni(
if start_str is not None and total > 0 and projection > 9:
now = datetime.datetime.now()
elapsed = (now - datetime.datetime.fromisoformat(start_str)).total_seconds()
# Exclude detected idle time (beamline-down pauses, a crash +
# restart gap, ...) so it doesn't make the scan look slower than
# it actually is while it's running.
elapsed -= self.progress.get("accumulated_idle_time", 0.0)
elapsed = max(elapsed, 1.0) # guard against a degenerate/negative denominator
rate = projection / elapsed # projections per second
remaining_s = (total - projection) / rate
self.progress["estimated_remaining_time"] = remaining_s
@@ -2060,8 +2335,8 @@ class Flomni(
print("\x1b[95mProgress report:")
print(f"Tomo type: ....................... {self.progress['tomo_type']}")
print(f"Projection: ...................... {self.progress['projection']:.0f}")
print(f"Total projections expected ....... {self.progress['total_projections']}")
print(f"Angle: ........................... {self.progress['angle']}")
print(f"Total projections expected ....... {self.progress['total_projections']:.0f}")
print(f"Angle: ........................... {self.progress['angle']:.3f}")
print(f"Current subtomo: ................. {self.progress['subtomo']}")
print(f"Current projection within subtomo: {self.progress['subtomo_projection']}")
print(f"Estimated remaining time: ........ {eta_str}")
@@ -2142,9 +2417,10 @@ class Flomni(
return angle, subtomo_number
def tomo_reconstruct(
self, base_path="~/data/raw/logs/reconstruction_queue", probe_propagation: float | None = None
self,
base_path="~/data/raw/logs/reconstruction_queue",
probe_propagation: float | None = None,
):
"""write the tomo reconstruct file for the reconstruction queue"""
bec = builtins.__dict__.get("bec")
@@ -2182,7 +2458,7 @@ class Flomni(
+ self.manual_shift_y
)
sum_offset_z = offsets[2]
#TODO this fix is while the tracker z is broken
# TODO this fix is while the tracker z is broken
probe_propagation = -sum_offset_z * 1e-6
sum_offset_z = 0
@@ -2273,20 +2549,40 @@ class Flomni(
dev.rtx.controller.move_samx_to_scan_region(sum_offset_x)
if tracker_signal == "low":
logger.warning(
"Signal strength of the laser tracker is low. Realignment recommended!"
)
logger.warning("Signal strength of the laser tracker is low. Realignment recommended!")
elif tracker_signal == "toolow":
raise FlomniError(
"Signal strength of the laser tracker is too low for scanning. Realignment required!"
)
# --- acquire ---
n_frames = (
frames_per_trigger if frames_per_trigger is not None else self.frames_per_trigger
)
n_frames = frames_per_trigger if frames_per_trigger is not None else self.frames_per_trigger
scans.acquire(exp_time=self.tomo_countingtime, frames_per_trigger=n_frames)
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):
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
total_projections: N * 8
self.tomo_angle_stepsize is stored as a raw, uncorrected value
(derived once from whatever total was originally typed in); reading
it back directly does not generally equal the achievable grid. Any
code that displays or prompts using these quantities should go
through this helper rather than recomputing the formula locally --
that duplication (display using a different formula than the one
sub_tomo_scan() actually uses) is exactly what previously caused
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
return N, step, N * 8
def tomo_parameters(self):
"""print and update the tomo parameters"""
print("Current settings:")
@@ -2303,11 +2599,20 @@ class Flomni(
if self.tomo_type == 1:
print("\x1b[1mTomo type 1:\x1b[0m 8 equally spaced sub-tomograms")
print(f"Angular range = {self.tomo_angle_range} degrees")
print(f"Total number of projections: {(self.tomo_angle_range/self.tomo_angle_stepsize)*8}")
print(f"Angular step within sub-tomogram: {self.tomo_angle_stepsize} degrees")
# N, step, total_projections all come from the same helper
# sub_tomo_scan() effectively uses internally - see
# _tomo_type1_actual_grid() for why this can't just read
# self.tomo_angle_stepsize directly.
_, 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")
print(
"Angular step of the final (combined) tomogram:"
f" {self.tomo_angle_range/((self.tomo_angle_range/self.tomo_angle_stepsize)*8)} degrees"
f" {self.tomo_angle_range / total_projections:.3f} degrees"
)
print(
"0-deg reference shots (odd sub-tomo start + end) ="
f" {self.zero_deg_reference_at_each_subtomo}"
)
if self.tomo_type == 2:
print("\x1b[1mTomo type 2:\x1b[0m Golden ratio tomography")
@@ -2324,8 +2629,8 @@ class Flomni(
print(
"\x1b[1mTomo type 3:\x1b[0m Equally spaced tomography, golden ratio starting angle"
)
print(f"Number of projections per sub-tomogram: {180/self.tomo_angle_stepsize}")
print(f"Angular step within sub-tomogram: {self.tomo_angle_stepsize} degrees")
print(f"Number of projections per sub-tomogram: {180 / self.tomo_angle_stepsize:.3f}")
print(f"Angular step within sub-tomogram: {self.tomo_angle_stepsize:.3f} degrees")
if self.golden_max_number_of_projections > 0:
print(f"ending after {self.golden_max_number_of_projections} projections")
else:
@@ -2356,10 +2661,10 @@ class Flomni(
self.ptycho_reconstruct_foldername = self._get_val(
"Reconstruction queue ", self.ptycho_reconstruct_foldername, str
)
self.manual_shift_y = self._get_val("<manual_shift_y> um", self.manual_shift_y, float)
self.frames_per_trigger = self._get_val(
"Frames per trigger (burst)", self.frames_per_trigger, int
)
self.manual_shift_y = self._get_val("<manual_shift_y> um", self.manual_shift_y, float)
self.single_point_instead_of_fermat_scan = bool(
self._get_val(
"Single point instead of fermat scan (acquire at angle) 1/0?",
@@ -2387,18 +2692,49 @@ class Flomni(
self.tomo_angle_range = self._get_val(
"Angular range (180 or 360)", self.tomo_angle_range, int
)
# Default shown here must be the actual achievable total
# (int(range/stepsize)*8), not the raw algebraic inverse of
# the stored stepsize -- the latter just reconstructs
# whatever total was typed in on a PREVIOUS call (e.g. 63),
# even if that got silently adjusted to 56 at the time; this
# is also what was causing the next default shown here to
# never reflect a prior adjustment.
_, _, current_total = self._tomo_type1_actual_grid()
tomo_numberofprojections = self._get_val(
"Total number of projections",
(self.tomo_angle_range / self.tomo_angle_stepsize) * 8,
int,
"Total number of projections", current_total, int
)
self.tomo_angle_stepsize = (self.tomo_angle_range / tomo_numberofprojections) * 8
# Now report what was ACTUALLY achieved, via the same helper
# sub_tomo_scan() effectively uses -- not the raw value just
# typed in or the raw stored stepsize, either of which can
# silently disagree with what the scan will actually run
# (this was previously the case: e.g. requesting 63 stored a
# stepsize whose raw value differed from the achievable
# per-projection step the scan actually used, 22.857 vs
# 25.714 in one observed case).
_, achievable_step, actual_total = self._tomo_type1_actual_grid()
if actual_total != tomo_numberofprojections:
print(
f"Note: {tomo_numberofprojections} projections does not divide evenly "
f"into 8 equally spaced sub-tomograms over {self.tomo_angle_range} "
f"degrees; adjusted to the nearest achievable total of {actual_total} "
"projections to keep the angular grid uniform."
)
print(
f"The angular step within a sub-tomogram will be {self.tomo_angle_stepsize} degrees"
f"The angular step within a sub-tomogram will be {achievable_step:.3f} degrees"
)
print(
"The angular step of the final (combined) tomogram will be"
f" {self.tomo_angle_range / tomo_numberofprojections} degrees"
f" {self.tomo_angle_range / actual_total:.3f} degrees"
)
self.zero_deg_reference_at_each_subtomo = bool(
self._get_val(
"Take 0-deg reference shots (start of each odd sub-tomo + end) for"
" damage estimation 1/0?",
int(self.zero_deg_reference_at_each_subtomo),
int,
)
)
if self.tomo_type == 2:
@@ -2440,6 +2776,177 @@ class Flomni(
def _get_val(msg: str, default_value, data_type):
return data_type(input(f"{msg} ({default_value}): ") or default_value)
# Ordered set of all global-var-backed tomo scan parameters: exactly the
# settings shown by tomo_parameters(), plus manual_shift_y/
# tomo_stitch_overlap/corridor_size which also affect the scan but are
# only set directly as properties. This is the full "parameter set" that
# tomo_queue_add()/tomo_queue_execute() snapshot and restore.
_TOMO_QUEUE_PARAM_NAMES = (
"tomo_countingtime",
"tomo_shellstep",
"fovx",
"fovy",
"stitch_x",
"stitch_y",
"tomo_stitch_overlap",
"ptycho_reconstruct_foldername",
"manual_shift_y",
"frames_per_trigger",
"single_point_instead_of_fermat_scan",
"tomo_type",
"tomo_angle_range",
"tomo_angle_stepsize",
"golden_ratio_bunch_size",
"golden_max_number_of_projections",
"golden_projections_at_0_deg_for_damage_estimation",
"zero_deg_reference_at_each_subtomo",
"corridor_size",
)
def tomo_queue_add(self, label: str = None) -> int:
"""Snapshot the currently set tomo parameters and append them as a
new job to the tomo queue (persisted, survives a kernel restart).
Typical usage::
flomni.tomo_parameters() # set up parameter set #1
flomni.tomo_queue_add("fast overview scan")
flomni.tomo_parameters() # change to parameter set #2
flomni.tomo_queue_add("hires scan")
flomni.tomo_queue_show()
flomni.tomo_queue_execute() # runs both, in order, on this sample
Args:
label: Optional name for the job, shown by tomo_queue_show().
Defaults to "job_<n>".
Returns:
The index of the newly added job.
"""
params = {name: getattr(self, name) for name in self._TOMO_QUEUE_PARAM_NAMES}
index = len(self._tomo_queue_proxy)
job = {
"label": label or f"job_{index + 1}",
"params": params,
"status": "pending",
"added_at": datetime.datetime.now().isoformat(),
}
self._tomo_queue_proxy.append(job)
print(f"Added tomo queue job #{index} ({job['label']}).")
return index
def tomo_queue_delete(self, *indices: int) -> None:
"""Delete one or more jobs from the tomo queue by index.
Accepts any number of indices, e.g. ``flomni.tomo_queue_delete(2, 5)``
to drop several jobs in one call. All indices are resolved against
the queue as it currently stands and deleted highest-index-first, so
passing several indices in any order is safe and won't shift the
meaning of the indices still to be deleted.
"""
if not indices:
print("No index given.")
return
for index in sorted(set(indices), reverse=True):
job = self._tomo_queue_proxy.pop(index)
print(f"Deleted tomo queue job #{index} ({job['label']}).")
def tomo_queue_clear(self) -> None:
"""Empty the tomo queue."""
self._tomo_queue_proxy.clear()
print("Tomo queue cleared.")
def tomo_queue_show(self) -> list:
"""Print and return the current tomo queue, one line per job."""
jobs = self._tomo_queue_proxy.as_list()
if not jobs:
print("Tomo queue is empty.")
return jobs
for i, job in enumerate(jobs):
p = job["params"]
print(
f"[{i}] {job['status']:>10s} {job['label']:<20s} "
f"type={p['tomo_type']} fov={p['fovx']}/{p['fovy']}um "
f"step={p['tomo_shellstep']}um ctime={p['tomo_countingtime']}s "
f"range={p['tomo_angle_range']}deg"
)
return jobs
def tomo_queue_execute(self, start_index: int = 0) -> None:
"""Run all pending tomo queue jobs in sequence, on the current sample.
For each job, restores its snapshotted parameters onto the live
properties (exactly as if set by hand) and then calls
``tomo_scan()`` -- or, for a job that didn't run to completion last
time, ``tomo_scan_resume()``, so it picks back up mid-scan instead
of restarting from subtomo 1 / angle 0. Jobs already marked "done"
are skipped on the next call, so simply calling tomo_queue_execute()
again resumes from where it stopped (e.g. after fixing a hardware
issue).
A job is considered not-yet-complete (and so gets resumed rather
than restarted) if its status is "incomplete" (a Python exception
was caught and execution stopped cleanly) OR "running" (the queue
process itself died -- killed kernel, dropped connection, power
loss, ... -- before it had a chance to record anything; in that
case the status is whatever was last written, which is "running",
not "incomplete", since the except block below never got to run).
Without treating "running" as resumable too, a real crash would
cause this method to silently restart that job from scratch on the
next call instead of resuming it.
If you've already manually called flomni.tomo_scan_resume()
yourself to recover from a crash (bypassing the queue), that scan
is now actually finished even though the queue still has the job
marked "incomplete" or "running" -- mark it done yourself before
calling this again, or it will be re-run from scratch:
flomni._tomo_queue_proxy.update(job_index, status="done")
Args:
start_index: Queue index to start from. Defaults to 0, but jobs
already marked "done" are skipped automatically either way.
"""
jobs = self._tomo_queue_proxy.as_list()
if not jobs:
print("Tomo queue is empty.")
return
if not self.OMNYTools.yesno(
f"Starting automatic execution of {len(jobs) - start_index} queued tomo scan(s) on"
f" sample '{self.sample_name}'. OK?",
"y",
):
print("Aborted.")
return
for i in range(start_index, len(jobs)):
job = jobs[i]
if job["status"] == "done":
continue
resume_job = job["status"] in ("incomplete", "running")
print(f"\n=== Tomo queue job {i + 1}/{len(jobs)}: {job['label']} ===")
for name, value in job["params"].items():
setattr(self, name, value)
self._tomo_queue_proxy.update(i, status="running")
try:
if resume_job:
self.tomo_scan_resume()
else:
self.tomo_scan()
except Exception as exc:
self._tomo_queue_proxy.update(i, status="incomplete")
print(f"Tomo queue job {i} ({job['label']}) did not complete: {exc}")
print(
"Queue paused. Fix the issue and call tomo_queue_execute() "
"again to resume from this job."
)
raise
self._tomo_queue_proxy.update(i, status="done")
print("\nTomo queue finished -- all jobs done.")
def rt_off(self):
dev.rtx.enabled = False
dev.rty.enabled = False
@@ -2540,4 +3047,4 @@ if __name__ == "__main__":
builtins.__dict__["bec"] = bec
builtins.__dict__["umv"] = umv
flomni = Flomni(bec)
flomni.start_x_ray_eye_alignment()
flomni.start_x_ray_eye_alignment()