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

feat(flomni): log per-projection and per-tomogram scan timing AND guard tomo_scan_projection against single-point mode

Browse Source 
Add JSONL timing logs under ~/data/raw/logs/timing_statistics/ for
building a scan-time prediction model. Times each projection in
_tomo_scan_at_angle (final successful attempt only, retries excluded)
and each tomogram in tomo_scan, capturing raw scan parameters plus
elapsed/idle/active durations.

tomo_scan_projection always runs a fermat scan; warn and confirm when
called directly with single_point_instead_of_fermat_scan set, pointing
to tomo_acquire_at_angle instead. Internal callers pass _internal=True
to skip the prompt.
This commit is contained in:
x12sa
2026-07-07 10:41:40 +02:00
committed by holler
co-authored by holler
parent 5a12465ab1
commit c55e5c8e73
1 changed files with 150 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+150 -3
View File
@@ -1,5 +1,6 @@
import builtins
import datetime
import json
import os
import random
import subprocess
@@ -1867,7 +1868,7 @@ class Flomni(
while not successful:
try:
start_scan_number = bec.queue.next_scan_number
self.tomo_scan_projection(angle)
self.tomo_scan_projection(angle, _internal=True)
error_caught = False
except AlarmBase as exc:
if exc.alarm_type == "TimeoutError":
@@ -2103,13 +2104,25 @@ class Flomni(
else:
num_repeats = 1
start_scan_number = bec.queue.next_scan_number
projection_start = time.perf_counter()
for i in range(num_repeats):
self._at_each_angle(angle)
projection_duration = time.perf_counter() - projection_start
end_scan_number = bec.queue.next_scan_number
for scan_nr in range(start_scan_number, end_scan_number):
self._write_tomo_scan_number(scan_nr, angle, subtomo_number)
# Only reached when the projection completed without @scan_repeat
# re-raising, so this is the final successful attempt's duration.
self._log_projection_timing(
angle=angle,
subtomo_number=subtomo_number,
duration_s=projection_duration,
start_scan_number=start_scan_number,
end_scan_number=end_scan_number,
)
def tomo_scan(self, subtomo_start=1, start_angle=None, projection_number=None):
"""start a tomo scan"""
@@ -2284,6 +2297,7 @@ class Flomni(
self.progress["projection"] = self.progress["total_projections"]
self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
self._print_progress()
self._log_tomogram_timing()
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))}"
@@ -2409,7 +2423,7 @@ class Flomni(
self.tomo_acquire_at_angle(angle)
return
self.tomo_scan_projection(angle)
self.tomo_scan_projection(angle, _internal=True)
def _golden(self, ii, howmany_sorted, maxangle, reverse=False):
"""returns the iis golden ratio angle of sorted bunches of howmany_sorted and its subtomo number"""
@@ -2477,6 +2491,117 @@ class Flomni(
random_offset_y=random_offset_y,
)
_TIMING_LOG_DIR = "~/data/raw/logs/timing_statistics"
_TIMING_SETUP = "flomni"
_PROJECTION_TIMING_LOG = "projection_timing_log.jsonl"
_TOMOGRAM_TIMING_LOG = "tomogram_timing_log.jsonl"
def _append_timing_record(self, filename: str, record: dict) -> None:
"""Append one JSON record as a line to a timing-statistics log file.
Deliberately best-effort: a failure to write a timing record must
never abort or interfere with an in-progress measurement, so any
exception here is caught and logged rather than propagated. The
files live in a dedicated subfolder (self._TIMING_LOG_DIR), separate
from tomography_scannumbers.txt and any other existing log, and are
pure append-only JSONL (one JSON object per line) so they are
trivial to load later with pandas.read_json(..., lines=True) for the
eventual scan-time prediction model.
"""
try:
log_dir = os.path.expanduser(self._TIMING_LOG_DIR)
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, filename)
with open(log_file, "a+") as out_file:
out_file.write(json.dumps(record) + "\n")
except Exception as exc: # pylint: disable=broad-except
logger.warning(f"Failed to write timing record to {filename}: {exc}")
def _log_projection_timing(
self,
angle: float,
subtomo_number: int,
duration_s: float,
start_scan_number: int,
end_scan_number: int,
) -> None:
"""Write one per-projection timing record.
A "projection" here is one completed _at_each_angle() call (all the
stitch tiles / the single-point acquisition for one angle). Only
successfully completed projections reach this point: because
_tomo_scan_at_angle is wrapped in @scan_repeat, a failed attempt
re-invokes the whole method and never returns to the logging call,
so retried attempts are naturally excluded and only the final
successful duration is recorded.
The parameters logged are exactly the scan settings that plausibly
drive the duration (FOV, step, counting time, burst frames, stitch
tiling, corridor, single-point mode) -- the raw feature set for the
prediction model. The actual fermat-scan point count is deliberately
not resolved here; if it turns out to be needed it can be added
later once real data shows the raw settings aren't enough.
"""
record = {
"setup": self._TIMING_SETUP,
"timestamp": datetime.datetime.now().isoformat(),
"duration_s": duration_s,
"angle": angle,
"subtomo_number": subtomo_number,
"start_scan_number": start_scan_number,
"end_scan_number": end_scan_number,
"n_scans": end_scan_number - start_scan_number,
"tomo_type": self.tomo_type,
"single_point_instead_of_fermat_scan": self.single_point_instead_of_fermat_scan,
"fovx": self.fovx,
"fovy": self.fovy,
"tomo_shellstep": self.tomo_shellstep,
"tomo_countingtime": self.tomo_countingtime,
"frames_per_trigger": self.frames_per_trigger,
"stitch_x": self.stitch_x,
"stitch_y": self.stitch_y,
"tomo_stitch_overlap": self.tomo_stitch_overlap,
"corridor_size": self.corridor_size,
}
self._append_timing_record(self._PROJECTION_TIMING_LOG, record)
def _log_tomogram_timing(self) -> None:
"""Write one per-tomogram timing record at the end of a tomo_scan().
Captures a full snapshot of the scan parameters (reusing
_TOMO_QUEUE_PARAM_NAMES -- the single source of truth for "every
parameter that affects how the scan runs") together with the
tomogram-level wall-clock timing already tracked in self.progress:
total elapsed, the accumulated idle time detected from inter-
projection gaps, and the active (elapsed-minus-idle) measurement
time. Standalone by design -- no tomo_id / sample-database cross-
reference for now.
"""
start_str = self.progress.get("tomo_start_time")
now = datetime.datetime.now()
elapsed_s = None
if start_str is not None:
try:
elapsed_s = (now - datetime.datetime.fromisoformat(start_str)).total_seconds()
except (ValueError, TypeError):
elapsed_s = None
idle_s = self.progress.get("accumulated_idle_time", 0.0)
active_s = elapsed_s - idle_s if elapsed_s is not None else None
record = {
"setup": self._TIMING_SETUP,
"timestamp": now.isoformat(),
"tomo_start_time": start_str,
"elapsed_s": elapsed_s,
"accumulated_idle_time_s": idle_s,
"active_s": active_s,
"total_projections": self.progress.get("total_projections"),
"tomo_type_label": self.progress.get("tomo_type"),
"params": {name: getattr(self, name) for name in self._TOMO_QUEUE_PARAM_NAMES},
}
self._append_timing_record(self._TOMOGRAM_TIMING_LOG, record)
def _write_tomo_scan_number(self, scan_number: int, angle: float, subtomo_number: int) -> None:
tomo_scan_numbers_file = os.path.expanduser("~/data/raw/logs/tomography_scannumbers.txt")
with open(tomo_scan_numbers_file, "a+") as out_file:
@@ -2485,7 +2610,29 @@ class Flomni(
f"{scan_number} {angle} {dev.fsamroy.read()['fsamroy']['value']:.5f} {self.tomo_id} {subtomo_number} {0} {self.sample_name}\n"
)
def tomo_scan_projection(self, angle: float):
def tomo_scan_projection(self, angle: float, _internal: bool = False):
"""Acquire one fermat-scan projection at `angle`.
Note: this ALWAYS runs a fermat scan, regardless of
single_point_instead_of_fermat_scan. That flag is honored by the
normal tomo flow (tomo_scan -> _at_each_angle) and by
tomo_acquire_at_angle, not here -- a single named command does one
kind of acquisition. If you call this directly at the CLI with the
single-point flag set, you probably meant tomo_acquire_at_angle();
the guard below points that out. Internal callers that legitimately
want the fermat path regardless (the fermat branch of
_at_each_angle, and tomo_alignment_scan, which needs real ptycho
data) pass _internal=True to skip the prompt.
"""
if not _internal and self.single_point_instead_of_fermat_scan:
print(
"\x1b[93mWarning: single_point_instead_of_fermat_scan is set, but"
" tomo_scan_projection always runs a FERMAT scan. For a single-point"
" acquisition at this angle use tomo_acquire_at_angle(angle) instead.\x1b[0m"
)
if not self.OMNYTools.yesno("Run a fermat scan anyway?", "n"):
print("Aborted. Use tomo_acquire_at_angle(angle) for a single-point acquisition.")
return
dev.rtx.controller.laser_tracker_check_signalstrength()