From e59a42f97868eb15e373b0872bca6c3a27a74ec3 Mon Sep 17 00:00:00 2001 From: x01dc Date: Tue, 21 Jul 2026 10:50:08 +0200 Subject: [PATCH] feat/port timing-statistics log + scilog_last_ptycho_scans() to lamni Ports flomni's per-projection/tomogram JSONL timing log (~/data/raw/logs/timing_statistics/*.jsonl, feeding a future scan-time prediction model) and the scilog_last_ptycho_scans(n) user command to lamni. Direct port adapted only for lamni's own param names (no fovx/fovy/stitch_x/stitch_y/single_point_instead_of_fermat_scan -- uses tomo_circfov/lamni_stitch_x/y/lamni_piezo_range_x/y instead, no single-point concept at all). _log_projection_timing() is only called on a clean (non-error_caught) completion of _tomo_scan_at_angle()'s retry loop, so an AlarmBase/TimeoutError attempt (no reliable duration measurement) doesn't pollute the log. Item 1 of csaxs_bec/bec_ipython_client/plugins/LamNI/AI_docs/ FLOMNI_LAMNI_FEATURE_GAPS_2026-07.md. Co-Authored-By: Claude Sonnet 5 --- .../bec_ipython_client/plugins/LamNI/lamni.py | 227 ++++++++++++++++++ .../test_lamni_timing_log.py | 136 +++++++++++ 2 files changed, 363 insertions(+) create mode 100644 tests/tests_bec_ipython_client/test_lamni_timing_log.py diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py index 6e951b7..93a4e05 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py @@ -1,5 +1,6 @@ import builtins import datetime +import json import os import subprocess import time @@ -714,6 +715,216 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools tags, ) + _TIMING_LOG_DIR = "~/data/raw/logs/timing_statistics" + _TIMING_SETUP = "lamni" + _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 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 (piezo range, step, counting time, burst frames, + stitch tiling, circular FOV crop, corridor) -- the raw feature set + for the prediction model. Mirrors Flomni._log_projection_timing() + (fovx/fovy/stitch_x/stitch_y/single_point_instead_of_fermat_scan + there become lamni_piezo_range_x/y/lamni_stitch_x/y/tomo_circfov + here -- lamni has no single-point acquisition mode at all). + """ + 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, + "lamni_piezo_range_x": self.lamni_piezo_range_x, + "lamni_piezo_range_y": self.lamni_piezo_range_y, + "tomo_circfov": self.tomo_circfov, + "tomo_shellstep": self.tomo_shellstep, + "tomo_countingtime": self.tomo_countingtime, + "frames_per_trigger": self.frames_per_trigger, + "lamni_stitch_x": self.lamni_stitch_x, + "lamni_stitch_y": self.lamni_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_SCAN_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. Mirrors Flomni._log_tomogram_timing() exactly. + """ + 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_SCAN_PARAM_NAMES}, + } + self._append_timing_record(self._TOMOGRAM_TIMING_LOG, record) + + def _read_last_timing_records(self, number_of_scans: int) -> list[dict]: + """Return the last ``number_of_scans`` projection timing records. + + Reads back the append-only projection timing log written by + _log_projection_timing(). Returns a list ordered oldest-to-newest + (i.e. in acquisition order), or an empty list if the log is missing + or unreadable. + """ + log_file = os.path.join( + os.path.expanduser(self._TIMING_LOG_DIR), self._PROJECTION_TIMING_LOG + ) + records: list[dict] = [] + try: + with open(log_file, "r") as in_file: + lines = [ln for ln in in_file if ln.strip()] + for line in lines[-number_of_scans:]: + try: + records.append(json.loads(line)) + except json.JSONDecodeError: + continue + except FileNotFoundError: + print(f"scilog_last_scans: timing log not found at {log_file}") + except Exception as exc: # pylint: disable=broad-except + print(f"scilog_last_scans: could not read timing log: {exc}") + return records + + def scilog_last_ptycho_scans(self, number_of_scans: int = 1) -> None: + """Write a scilog entry with a user comment and recent scan info. + + The user is prompted for a free-text comment, which appears at the + top of the entry. For each of the last ``number_of_scans`` projections + (1 to 4) the entry then lists the scan number(s), field of view + (piezo range), exposure (counting) time and scan duration, read from + the projection timing log. Pass ``number_of_scans=0`` to send a + comment only, with no scan information. + + Only ptycho projections write a timing record, so non-ptycho scans do + not appear; if fewer records exist than requested, whatever is + available is used and a note is added. + """ + if not isinstance(number_of_scans, int) or isinstance(number_of_scans, bool): + print("scilog_last_scans: number_of_scans must be an integer between 0 and 4.") + return + if not 0 <= number_of_scans <= 4: + print("scilog_last_scans: number_of_scans must be between 0 and 4.") + return + + records = self._read_last_timing_records(number_of_scans) if number_of_scans else [] + if number_of_scans and not records: + print( + "scilog_last_scans: no scans found in the timing log; " + "sending a comment-only entry." + ) + + comment = input("Enter a comment for the scilog entry: ").strip() + if not comment and not records: + print("scilog_last_scans: empty comment and no scans — nothing to write.") + return + + lines = [] + if comment: + lines.append(f"{comment}") + + if records: + if comment: + lines.append("") + if len(records) < number_of_scans: + lines.append( + f"LamNI summary of the last {len(records)} scan(s) " + f"(only {len(records)} of {number_of_scans} requested were found):" + ) + else: + lines.append("LamNI parameters:") + lines.append("") + for rec in records: + start = rec.get("start_scan_number") + end = rec.get("end_scan_number") + if start is not None and end is not None and end - start > 1: + scan_str = f"{start}-{end - 1}" + elif start is not None: + scan_str = f"{start}" + else: + scan_str = "?" + fovx = rec.get("lamni_piezo_range_x") + fovy = rec.get("lamni_piezo_range_y") + fov_str = f"{fovx} x {fovy} um" if fovx is not None and fovy is not None else "?" + exposure = rec.get("tomo_countingtime") + exp_str = f"{exposure} s" if exposure is not None else "?" + duration = rec.get("duration_s") + dur_str = self._format_duration(duration) if duration is not None else "?" + lines.append( + f"scan {scan_str}: FOV {fov_str}, exposure {exp_str}, duration {dur_str}" + ) + + content = "\n".join(lines) + print(content) + + self.write_to_scilog(content, ["tomoscan"]) + 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: @@ -1034,8 +1245,10 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools num_repeats = 1 try: 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 error_caught = False except AlarmBase as exc: if exc.alarm_type == "TimeoutError": @@ -1049,6 +1262,19 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools for scan_nr in range(start_scan_number, end_scan_number): self._write_tomo_scan_number(scan_nr, angle, subtomo_number) + if not error_caught: + # Only reached for a clean completion -- an + # AlarmBase/TimeoutError attempt above has no reliable + # duration measurement, so it's excluded from the + # timing log rather than polluting it. + 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, + ) + successful = True def _golden(self, ii, howmany_sorted, maxangle=360, reverse=False): @@ -1267,6 +1493,7 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools 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") idle_s = self.progress.get("accumulated_idle_time", 0.0) diff --git a/tests/tests_bec_ipython_client/test_lamni_timing_log.py b/tests/tests_bec_ipython_client/test_lamni_timing_log.py new file mode 100644 index 0000000..e3a0da1 --- /dev/null +++ b/tests/tests_bec_ipython_client/test_lamni_timing_log.py @@ -0,0 +1,136 @@ +"""Tests for lamni's timing-statistics log + scilog_last_ptycho_scans(), +ported from flomni (see csaxs_bec/bec_ipython_client/plugins/LamNI/AI_docs/ +FLOMNI_LAMNI_FEATURE_GAPS_2026-07.md, item 1). + +Uses a bare LamNI instance (bypassing __init__'s heavy side effects), same +pattern as test_lamni_tomo_angles.py/test_fermat_position_warning.py. +""" + +import json + +import pytest + +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI, _ProgressProxy + + +class FakeClient: + """Minimal in-memory stand-in for BEC's global-var store.""" + + def __init__(self): + self._vars = {} + + def get_global_var(self, key): + return self._vars.get(key) + + def set_global_var(self, key, value): + self._vars[key] = value + + +def make_lamni(tmp_path): + obj = object.__new__(LamNI) + obj.client = FakeClient() + obj._progress_proxy = _ProgressProxy(obj.client) + obj.tomo_id = -1 + obj._TIMING_LOG_DIR = str(tmp_path) + return obj + + +def _read_jsonl(path): + with open(path) as f: + return [json.loads(line) for line in f if line.strip()] + + +def test_append_and_read_timing_records_round_trip(tmp_path): + lamni = make_lamni(tmp_path) + lamni._append_timing_record(lamni._PROJECTION_TIMING_LOG, {"a": 1}) + lamni._append_timing_record(lamni._PROJECTION_TIMING_LOG, {"a": 2}) + lamni._append_timing_record(lamni._PROJECTION_TIMING_LOG, {"a": 3}) + + records = lamni._read_last_timing_records(2) + + assert [r["a"] for r in records] == [2, 3] + + +def test_read_timing_records_missing_file_returns_empty(tmp_path, capsys): + lamni = make_lamni(tmp_path) + records = lamni._read_last_timing_records(2) + assert records == [] + assert "timing log not found" in capsys.readouterr().out + + +def test_append_timing_record_tolerates_write_failure(tmp_path, monkeypatch): + """Must never raise -- a timing-log write failure must not abort or + interfere with an in-progress measurement.""" + lamni = make_lamni(tmp_path) + lamni._TIMING_LOG_DIR = "/nonexistent-root-that-cannot-be-created/xyz" + lamni._append_timing_record(lamni._PROJECTION_TIMING_LOG, {"a": 1}) # must not raise + + +def test_log_projection_timing_uses_lamni_param_names(tmp_path): + lamni = make_lamni(tmp_path) + lamni.tomo_type = 1 + lamni.tomo_shellstep = 1.0 + lamni.tomo_countingtime = 0.1 + lamni.frames_per_trigger = 1 + lamni.lamni_stitch_x = 1 + lamni.lamni_stitch_y = 0 + lamni.tomo_stitch_overlap = 0.2 + lamni.corridor_size = -1 + lamni.client.set_global_var("lamni_piezo_range_x", 20.0) + lamni.client.set_global_var("lamni_piezo_range_y", 20.0) + lamni.tomo_circfov = 0.0 + + lamni._log_projection_timing( + angle=12.5, subtomo_number=1, duration_s=3.2, start_scan_number=10, end_scan_number=12 + ) + + records = _read_jsonl(f"{tmp_path}/{lamni._PROJECTION_TIMING_LOG}") + assert len(records) == 1 + record = records[0] + assert record["setup"] == "lamni" + assert record["angle"] == 12.5 + assert record["n_scans"] == 2 + assert record["lamni_piezo_range_x"] == 20.0 + assert record["lamni_piezo_range_y"] == 20.0 + assert record["lamni_stitch_x"] == 1 + # lamni has no fovx/fovy/single_point_instead_of_fermat_scan concept + assert "fovx" not in record + assert "single_point_instead_of_fermat_scan" not in record + + +def test_log_tomogram_timing_snapshots_all_param_names(tmp_path): + lamni = make_lamni(tmp_path) + lamni.progress["tomo_start_time"] = None + lamni.progress["accumulated_idle_time"] = 0.0 + lamni.progress["total_projections"] = 144 + lamni.progress["tomo_type"] = "Equally spaced sub-tomograms" + + lamni._log_tomogram_timing() + + records = _read_jsonl(f"{tmp_path}/{lamni._TOMOGRAM_TIMING_LOG}") + assert len(records) == 1 + record = records[0] + assert record["setup"] == "lamni" + assert record["total_projections"] == 144 + assert set(record["params"]) == set(LamNI._TOMO_SCAN_PARAM_NAMES) + + +@pytest.mark.parametrize("bad_value", [-1, 5, 1.5, True, "2"]) +def test_scilog_last_ptycho_scans_rejects_invalid_number_of_scans(tmp_path, bad_value, capsys): + lamni = make_lamni(tmp_path) + lamni.scilog_last_ptycho_scans(bad_value) + out = capsys.readouterr().out + assert "must be" in out + + +def test_scilog_last_ptycho_scans_comment_only(tmp_path, monkeypatch): + lamni = make_lamni(tmp_path) + lamni.sample_name = "test" + sent = {} + lamni.write_to_scilog = lambda content, tags: sent.update(content=content, tags=tags) + monkeypatch.setattr("builtins.input", lambda *_: "test comment") + + lamni.scilog_last_ptycho_scans(0) + + assert sent["content"] == "test comment" + assert sent["tags"] == ["tomoscan"]