From c55e5c8e73cbe99481ed81b892a5dd7d981b5e5a Mon Sep 17 00:00:00 2001 From: x12sa Date: Fri, 3 Jul 2026 15:05:55 +0200 Subject: [PATCH] feat(flomni): log per-projection and per-tomogram scan timing AND guard tomo_scan_projection against single-point mode 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. --- .../plugins/flomni/flomni.py | 153 +++++++++++++++++- 1 file changed, 150 insertions(+), 3 deletions(-) diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py index d34af24..d9f2f8b 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py @@ -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()