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 <noreply@anthropic.com>
This commit is contained in:
x01dc
@@ -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)
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user