feat(flomni): comment-first scilog_last_scans, comment-only mode, no logo
scilog_last_scans now puts the user comment at the top of the entry and drops the flOMNI logo attachment. number_of_scans accepts 0 to send a comment-only entry. When fewer projection records exist than requested (e.g. non-ptycho scans, which write no timing record), the available ones are used and a note reports how many of the requested count were found.
This commit is contained in:
@@ -2969,6 +2969,108 @@ class Flomni(
|
||||
}
|
||||
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_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,
|
||||
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: {comment}")
|
||||
|
||||
if records:
|
||||
if comment:
|
||||
lines.append("")
|
||||
if len(records) < number_of_scans:
|
||||
lines.append(
|
||||
f"flOMNI summary of the last {len(records)} scan(s) "
|
||||
f"(only {len(records)} of {number_of_scans} requested were found):"
|
||||
)
|
||||
else:
|
||||
lines.append(f"flOMNI summary of the last {len(records)} scan(s):")
|
||||
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("fovx")
|
||||
fovy = rec.get("fovy")
|
||||
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)
|
||||
|
||||
bec = builtins.__dict__.get("bec")
|
||||
bec.messaging.scilog.new().add_text(content.replace("\n", "<br>")).add_tags(
|
||||
"tomoscan"
|
||||
).send()
|
||||
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user