feat(omny): add third progress ring, idle-time tracking, timing summary
CI for csaxs_bec / test (push) Failing after 5s

Mirko: "the flomni progress bar has three rings. here i am seing only
two... at the end of a tomogram in flomni a summary of the measurement
time and time lost is displayed. omny seems not to have that, maybe not
even measuring the time."

Both traced back to the same root gap left by the Phase-1 tomo-queue port:
nothing computed estimated_remaining_time/estimated_finish_time, and
nothing refreshed progress["heartbeat"] *during* a running scan (only
cleared it to None on completion) -- so accumulated_idle_time stayed 0.0
forever and there was no ETA data to show or summarize.

- _tomo_scan_at_angle() now refreshes the heartbeat every angle and
  attributes any gap beyond a normal-cadence heuristic to
  accumulated_idle_time (mirrors flomni's identical formula, minus its
  frames_per_trigger factor which OMNY has no property for).
- _print_progress() computes/stores/prints estimated_remaining_time and
  estimated_finish_time once the scan rate has stabilized.
- tomo_scan() prints an end-of-scan "Total measurement time"/"...excluding
  detected gaps"/"...lost to detected gaps" summary and sends it to scilog,
  direct port of flomni's block (minus its measured_log call, which OMNY
  has no equivalent of).
- gui_tools.py: omnygui_show_progress() adds flomni's third, scan-linked
  auto-updating ring; _omnygui_update_progress()'s center label now shows
  start time / ETA / estimated finish / active hook, matching flomni's
  _flomnigui_update_progress().

_describe_active_hook()/_active_hook_source() needed no porting -- already
provided by TomoQueueMixin since Phase 1. Not ported: flomni's separate
persistent timing-statistics-log subsystem (_log_tomogram_timing()) --
out of scope for what was actually asked (the printed summary + ring
count).

9 new tests (test_omny_tomo_scan.py, test_omny_gui_tools.py). Full suite:
731 passed.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QLrD7sVYGLAzsQjLVJpCgt
This commit is contained in:
x01dc
2026-09-03 02:45:59 +02:00
co-authored by Claude Sonnet 5
parent 52c1806fe2
commit afc13df24d
5 changed files with 376 additions and 37 deletions
@@ -1,4 +1,4 @@
# OMNY open issues (as of 2026-09-02)
# OMNY open issues (as of 2026-09-03)
Tracks the live-testing effort for bringing OMNY up to flomni's/BEC's current standards on
branch `update-omny`: bugs found (fixed and unfixed) and what's still untested. Written so a
@@ -26,14 +26,19 @@ removed" below. Tomo type 3 also gained a hard cap of 2.5 degrees on the angular
only a minimum-100-projections floor, which happened to already satisfy this but wasn't expressed
as a wear limit).
**Just landed**: the tomo-queue backend (Phase 1 of 2 of porting the tomo params GUI widget to
OMNY) — `OMNY` now has a real job queue, `at_each_angle_hook` registration, `tomo_scan_resume()`,
and scan progress visible outside the CLI session (previously a plain in-memory dict). See "Tomo-
queue backend ported to OMNY" below. Phase 2 (the widget itself) is a separate follow-up, not yet
started. Live-tested by Mirko against his own session (not a sim harness), which surfaced and
fixed several more real bugs — see "scilog writes hung / migrated to bec.messaging.scilog",
"tomo_scan() rejected TomoQueueMixin's interactive= kwarg", "stale ~/Data10/ paths crashed on a
fresh session", and "tomo_parameters() now warns about too few Fermat scan points" below.
**Just landed, confirmed working live**: the tomo-queue backend (Phase 1 of 2 of porting the tomo
params GUI widget to OMNY) — `OMNY` now has a real job queue, `at_each_angle_hook` registration,
`tomo_scan_resume()`, and scan progress visible outside the CLI session (previously a plain
in-memory dict). See "Tomo-queue backend ported to OMNY" below. Live-tested by Mirko against his
own session (not a sim harness), which surfaced and fixed several more real bugs along the way —
see "scilog writes hung / migrated to bec.messaging.scilog", "tomo_scan() rejected
TomoQueueMixin's interactive= kwarg", "stale ~/Data10/ paths crashed on a fresh session", and
"tomo_parameters() now warns about too few Fermat scan points" below. Mirko confirmed: "tomo scan
and tomo queue seems to work now for omny." Also just fixed following his feedback: the progress
ring GUI only had 2 of flomni's 3 rings, and OMNY never tracked/printed measurement time or idle
time the way flomni does at the end of a tomogram -- see "Progress ring GUI + ETA/idle-time
tracking + end-of-scan timing summary" below. Phase 2 (the `TomoParamsWidget` GUI itself) is a
separate follow-up, not yet started.
**Still open / deferred** (unchanged from before, not touched this pass):
- `x_ray_eye_align.py` is still LamNI-derived, unadapted code beyond the two call-site renames
@@ -150,19 +155,49 @@ Still untested (not attempted yet, not known-broken — just unexplored):
entry and its tomo-type selector/job-display logic is hardcoded to 3 tomo types (flomni's
and lamni's), not OMNY's 5 — needs its own plan once Phase 1 is live-tested.
## gui_tools.py progress ETA/heartbeat line (still not rendered, data now exists)
## Progress ring GUI + ETA/idle-time tracking + end-of-scan timing summary (fixed)
`gui_tools.py` was modernized to flomni's current single-level `.new()` / `_is_deleted()`-guard
pattern (window lifecycle, xray-eye-align GUI, cameras + hard-stop console, `RingProgressBar`).
As of the tomo-queue backend port (see below), `self.progress` now DOES carry
`tomo_start_time`/`heartbeat`/`estimated_remaining_time`/`estimated_finish_time` (global-var-backed,
same shape as flomni's/lamni's) — what's still not ported is purely a `gui_tools.py` rendering
detail: `_omnygui_update_progress()` was never updated to display an ETA/heartbeat line the way
flomni's `_flomnigui_update_progress()`/`_describe_active_hook()` do, and still only builds its
text from the original 7 fields. Not a crash risk either way — `_omnygui_update_progress()`
ignores fields it doesn't read, and `omny_webpage_generator.py` (unwired/dead code, see below)
already reads the ETA-ish fields defensively via `.get()`, degrading to "N/A" when absent. Revisit
whenever the progress ring GUI itself gets an ETA line, independent of the Phase 2 widget work.
Mirko live-tested Phase 1 and asked two related questions: why does OMNY's progress ring bar show
only 2 rings where flomni shows 3, and why does OMNY not print a measurement-time/time-lost summary
at the end of a tomogram the way flomni does. Both traced back to the same root gap: nothing in
OMNY ever computed `estimated_remaining_time`/`estimated_finish_time`, and nothing ever refreshed
`heartbeat` *during* a running scan (only cleared it to `None` on completion, from the Phase-1
port) -- so `accumulated_idle_time` stayed `0.0` forever and there was no ETA data to show or
summarize. All direct ports of flomni's already-working mechanism, all sharing the same underlying
data (`self.progress`, global-var-backed since Phase 1):
- **`_tomo_scan_at_angle()`** now refreshes `progress["heartbeat"]` at the start of every angle,
and -- if the gap since the previous heartbeat exceeds a normal-cadence heuristic
(`max(60s, 5 * tomo_countingtime)`, flomni's identical formula minus its `frames_per_trigger`
factor, which OMNY has no equivalent property for) -- attributes the excess to
`accumulated_idle_time` and prints a one-line note. This is what actually makes idle-time
tracking (and therefore both features below) possible; it didn't exist in Phase 1 at all.
- **`_print_progress()`** now computes and stores `estimated_remaining_time`/
`estimated_finish_time` from elapsed time minus accumulated idle time (once past projection 9,
matching flomni's stabilization threshold), and prints them alongside the existing progress
report.
- **`tomo_scan()`** now prints an end-of-scan summary ("Total measurement time", "... excluding
detected gaps", "... lost to detected gaps") after the try/finally, and sends the same content to
scilog via `write_to_scilog()` -- direct port of flomni's block, minus the flomni-only
`self.measured_log` sample-tracking call (OMNY has no equivalent).
- **`gui_tools.py`**: `omnygui_show_progress()` now adds a third ring with `set_update("scan")`
(auto-updated from the currently running scan, no manual `.set_value()` needed -- matches
flomni's `flomnigui_show_progress()` exactly), and `_omnygui_update_progress()`'s center-label
text now includes start time / ETA / estimated finish / active-hook lines, matching flomni's
`_flomnigui_update_progress()`.
`_format_duration()` (static, `"2h 03m 15s"`-style formatting) added to `omny.py` -- previously
flomni-only. `_describe_active_hook()`/`_active_hook_source()` needed no porting: already provided
by `TomoQueueMixin` since Phase 1.
**Not ported**: flomni's separate `_log_tomogram_timing()`/persistent timing-statistics-log
subsystem (writes a JSON record per tomogram to `~/data/raw/logs/timing_statistics`, backing
`scilog_last_ptycho_scans()`/a timing overview command) -- Mirko asked specifically about the
*printed* summary and the ring count, not this larger, separate logging feature. Flag if wanted.
Test coverage: 9 new tests (`test_omny_tomo_scan.py`: idle-time-from-gap, no-idle-time-within-
cadence, end-of-scan summary lines; `test_omny_gui_tools.py`: three-ring creation, center-label
text with/without an active hook). Full suite: 731 passed.
## cam_xeye device (simulated only, real hardware config still needed)
@@ -205,19 +205,26 @@ class OMNYGuiTools:
if self._omnygui_is_missing("progressbar"):
self.omnygui_remove_all_docks()
self.progressbar = self.gui.omny.new("RingProgressBar")
# Two manually-updated rings: overall tomogram progress and
# subtomo progress. Unlike flomni's third, auto-updating
# "scan"-linked ring, and unlike flomni's ETA/heartbeat/
# active-hook text, omny's progress dict doesn't carry that
# data yet (see omny/AI_docs/OPEN_ISSUES.md) -- kept simple
# here on purpose, revisit once that's ported.
# Two manually-updated rings (overall tomogram progress and
# subtomo progress) plus a third ring auto-updated from the
# currently running scan itself -- matches flomni's
# flomnigui_show_progress() exactly.
self.progressbar.add_ring().set_update("manual")
self.progressbar.add_ring().set_update("manual")
self.progressbar.add_ring().set_update("scan")
self._omnygui_update_progress()
except Exception as e:
logger.warning(f"Error in omnygui_show_progress: {e}")
def _omnygui_update_progress(self):
"""Update the progress ring bar and center label from the current progress state.
``self.progress`` is backed by the BEC global variable ``tomo_progress``
(see :class:`_ProgressProxy` in ``omny.py``), so this method reflects
the live state that is also accessible from other BEC client sessions via::
client.get_global_var("tomo_progress")
"""
try:
if self.progressbar is None:
logger.warning("OMNY progress GUI update skipped: progressbar is not available.")
@@ -233,6 +240,43 @@ class OMNYGuiTools:
main_progress_ring.set_value(progress)
subtomo_progress_ring.set_value(subtomo_progress)
# --- format start time for display --------------------------------
start_str = self.progress.get("tomo_start_time")
if start_str is not None:
import datetime as _dt
start_display = _dt.datetime.fromisoformat(start_str).strftime("%Y-%m-%d %H:%M:%S")
else:
start_display = "N/A"
# --- format estimated remaining time ------------------------------
remaining_s = self.progress.get("estimated_remaining_time")
if remaining_s is not None and remaining_s >= 0:
remaining_s = int(remaining_s)
h, rem = divmod(remaining_s, 3600)
m, s = divmod(rem, 60)
if h > 0:
eta_display = f"{h}h {m:02d}m {s:02d}s"
elif m > 0:
eta_display = f"{m}m {s:02d}s"
else:
eta_display = f"{s}s"
else:
eta_display = "N/A"
# ------------------------------------------------------------------
# --- format estimated finish (wall-clock) time --------------------
finish_str = self.progress.get("estimated_finish_time")
if finish_str is not None:
import datetime as _dt
finish_display = _dt.datetime.fromisoformat(finish_str).strftime(
"%Y-%m-%d %H:%M:%S"
)
else:
finish_display = "N/A"
# ------------------------------------------------------------------
text = (
f"Progress report:\n"
f" Tomo type: {self.progress['tomo_type']}\n"
@@ -241,8 +285,17 @@ class OMNYGuiTools:
f" Angle: {self.progress['angle']:.1f}\n"
f" Current subtomo: {self.progress['subtomo']}\n"
f" Current projection within subtomo: {self.progress['subtomo_projection']}\n"
f" Total projections per subtomo: {int(self.progress['subtomo_total_projections'])}"
f" Total projections per subtomo: {int(self.progress['subtomo_total_projections'])}\n"
f" Scan started: {start_display}\n"
f" Est. remaining: {eta_display}\n"
f" Est. finish: {finish_display}"
)
# self._describe_active_hook() comes from TomoQueueMixin (this
# is a mixin, always combined with it) -- shown only when a
# hook is actually active, same as tomo_parameters()'s CLI note.
hook_description = self._describe_active_hook()
if hook_description:
text += f"\n Hook: {hook_description}"
self.progressbar.set_center_label(text)
except Exception as exc:
logger.warning(f"OMNY progress GUI update skipped: {exc}")
@@ -1230,6 +1230,31 @@ class OMNY(
successful = False
error_caught = False
if 0 <= angle < 180.05:
now = datetime.datetime.now()
prev_heartbeat_str = self.progress.get("heartbeat")
if prev_heartbeat_str is not None:
gap = (now - datetime.datetime.fromisoformat(prev_heartbeat_str)).total_seconds()
# Normal cadence between consecutive projections is roughly the
# acquisition time plus motor/readout overhead. A gap well
# beyond that means something interrupted the scan in between
# (beamline-down interlock pause, a crash + manual restart,
# ...) -- attribute the excess to idle time so it doesn't drag
# down the apparent scan rate used for the ETA below. Mirrors
# flomni's identical heuristic (minus its frames_per_trigger
# factor -- OMNY has no such property). The 5x/60s margins
# are a heuristic, not a precise timing model -- tune if it
# over/under-triggers in practice.
normal_cadence = max(60.0, 5 * self.tomo_countingtime)
if gap > normal_cadence:
idle = gap - normal_cadence
self.progress["accumulated_idle_time"] = (
self.progress.get("accumulated_idle_time", 0.0) + idle
)
print(
f"Detected a {self._format_duration(gap)} gap since the last projection"
f" -- excluding {self._format_duration(idle)} from the ETA estimate."
)
self.progress["heartbeat"] = now.isoformat()
print(f"Starting OMNY scan for angle {angle} in subtomo {subtomo_number}")
self._print_progress()
self.omnygui_show_progress()
@@ -1436,6 +1461,47 @@ class OMNY(
# identical fix.
self.progress["heartbeat"] = None
self.progress["projection"] = self.progress["total_projections"]
self.progress["subtomo_projection"] = self.progress["subtomo_total_projections"]
self._print_progress()
self.OMNYTools.printgreenbold("Tomoscan finished")
idle_s = self.progress.get("accumulated_idle_time", 0.0)
start_str = self.progress.get("tomo_start_time")
elapsed_s = None
if start_str is not None:
try:
elapsed_s = (
datetime.datetime.now() - datetime.datetime.fromisoformat(start_str)
).total_seconds()
except (ValueError, TypeError):
elapsed_s = None
timing_lines = [
"Tomoscan finished.",
f"Measurement ID: {self.tomo_id}",
f"Sample: {self.sample_name}",
]
hook_description = self._describe_active_hook()
if hook_description:
timing_lines.append(f"At-each-angle hook: {hook_description}")
if elapsed_s is not None:
timing_lines.append(f"Total measurement time: {self._format_duration(elapsed_s)}")
timing_lines.append(
f"Total measurement time excluding detected gaps: {self._format_duration(elapsed_s - idle_s)}"
)
timing_lines.append(
f"Total measurement time lost to detected gaps: {self._format_duration(idle_s)}"
)
for line in timing_lines[3:]:
print(line)
timing_content = "\n".join(timing_lines)
hook_source = self._active_hook_source()
if hook_source:
timing_content += (
f"\n\nAt-each-angle hook source ('{self.at_each_angle_hook}'):\n{hook_source}"
)
self.write_to_scilog(timing_content, ["tomoscan"])
def tomo_scan_resume(self) -> None:
"""Resume a tomo_scan() that crashed or was interrupted, picking up
automatically from wherever ``progress`` last reported -- no need to
@@ -1501,7 +1567,42 @@ class OMNY(
start_angle = float(angles[projection_within_subtomo])
return subtomo_start, start_angle
@staticmethod
def _format_duration(seconds: float) -> str:
"""Format a duration in seconds as a human-readable string, e.g. '2h 03m 15s'."""
seconds = int(seconds)
h, remainder = divmod(seconds, 3600)
m, s = divmod(remainder, 60)
if h > 0:
return f"{h}h {m:02d}m {s:02d}s"
if m > 0:
return f"{m}m {s:02d}s"
return f"{s}s"
def _print_progress(self):
# --- compute and store estimated remaining time --------------------
start_str = self.progress.get("tomo_start_time")
projection = self.progress["projection"]
total = self.progress["total_projections"]
if start_str is not None and total > 0 and projection > 9:
now = datetime.datetime.now()
elapsed = (now - datetime.datetime.fromisoformat(start_str)).total_seconds()
# Exclude detected idle time (beamline-down pauses, a crash +
# restart gap, ...) so it doesn't make the scan look slower than
# it actually is while it's running.
elapsed -= self.progress.get("accumulated_idle_time", 0.0)
elapsed = max(elapsed, 1.0) # guard against a degenerate/negative denominator
rate = projection / elapsed # projections per second
remaining_s = (total - projection) / rate
self.progress["estimated_remaining_time"] = remaining_s
eta_str = self._format_duration(remaining_s)
finish_dt = now + datetime.timedelta(seconds=remaining_s)
self.progress["estimated_finish_time"] = finish_dt.isoformat()
finish_str = finish_dt.strftime("%Y-%m-%d %H:%M:%S")
else:
eta_str = "N/A"
finish_str = "N/A"
# ---------------------------------------------------------------------
print("\x1b[95mProgress report:")
print(f"Tomo type: ....................... {self.progress['tomo_type']}")
print(f"Projection: ...................... {self.progress['projection']:.0f}")
@@ -1509,9 +1610,9 @@ class OMNY(
print(f"Angle: ........................... {self.progress['angle']}")
print(f"Current subtomo: ................. {self.progress['subtomo']}")
print(f"Current projection within subtomo: {self.progress['subtomo_projection']}")
print(
f"Total projections per subtomo: ... {self.progress['subtomo_total_projections']}\x1b[0m"
)
print(f"Total projections per subtomo: ... {self.progress['subtomo_total_projections']}")
print(f"Estimated remaining time: ........ {eta_str}")
print(f"Estimated finish time: ........... {finish_str}\x1b[0m")
def add_sample_database(
self, samplename, date, eaccount, scan_number, setup, sample_additional_info, user
@@ -144,6 +144,78 @@ def test_omnygui_show_omnycam_parking_uses_current_image_widget_api():
assert tools.console is console
def _fake_progress(**overrides):
base = {
"tomo_type": "Equally spaced sub-tomograms (2)",
"projection": 3,
"total_projections": 10,
"angle": 12.5,
"subtomo": 1,
"subtomo_projection": 3,
"subtomo_total_projections": 5,
"tomo_start_time": None,
"estimated_remaining_time": None,
"estimated_finish_time": None,
}
base.update(overrides)
return base
def test_omnygui_show_progress_creates_three_rings_matching_flomni():
"""Third ring is auto-updated from the running scan itself
(set_update("scan")), matching flomni's flomnigui_show_progress() --
omny previously only had the two manually-updated rings."""
tools, client = _make_gui_tools()
client.gui.windows = {"omny": mock.MagicMock()}
progressbar = mock.MagicMock()
client.gui.omny.new.return_value = progressbar
ring1, ring2, ring3 = mock.MagicMock(), mock.MagicMock(), mock.MagicMock()
progressbar.add_ring.side_effect = [ring1, ring2, ring3]
progressbar.rings = [ring1, ring2, ring3]
tools.progress = _fake_progress()
tools._describe_active_hook = lambda: None
tools.omnygui_show_progress()
assert progressbar.add_ring.call_count == 3
ring1.set_update.assert_called_once_with("manual")
ring2.set_update.assert_called_once_with("manual")
ring3.set_update.assert_called_once_with("scan")
def test_omnygui_update_progress_text_includes_start_eta_finish():
tools, _ = _make_gui_tools()
tools.progressbar = mock.MagicMock()
tools.progressbar.rings = [mock.MagicMock(), mock.MagicMock()]
tools.progress = _fake_progress(
tomo_start_time="2026-01-01T00:00:00",
estimated_remaining_time=125.0,
estimated_finish_time="2026-01-01T01:00:00",
)
tools._describe_active_hook = lambda: None
tools._omnygui_update_progress()
text = tools.progressbar.set_center_label.call_args[0][0]
assert "Scan started: 2026-01-01 00:00:00" in text
assert "Est. remaining: 2m 05s" in text
assert "Est. finish: 2026-01-01 01:00:00" in text
assert "Hook:" not in text # no active hook
def test_omnygui_update_progress_shows_active_hook():
tools, _ = _make_gui_tools()
tools.progressbar = mock.MagicMock()
tools.progressbar.rings = [mock.MagicMock(), mock.MagicMock()]
tools.progress = _fake_progress()
tools._describe_active_hook = lambda: "my_hook"
tools._omnygui_update_progress()
text = tools.progressbar.set_center_label.call_args[0][0]
assert "Hook: my_hook" in text
def test_omnygui_show_omnycam_samplestage_uses_current_image_widget_api():
tools, client = _make_gui_tools()
client.gui.windows = {"omny": mock.MagicMock()}
@@ -15,6 +15,7 @@ exists in OMNY's own add_sample_database()).
"""
import builtins
import datetime
import types
import pytest
@@ -64,18 +65,27 @@ def make_omny_for_tomo_scan(monkeypatch, tomo_angle_stepsize: float, active_acco
obj._progress_proxy = _ProgressProxy(obj.client)
obj.tomo_type = 1
obj.tomo_angle_stepsize = tomo_angle_stepsize
obj.tomo_countingtime = 0.1
obj.tomo_id = 0
obj.special_angles = []
obj._current_special_angles = []
obj._at_each_angle_hooks = {} # normally set by _init_tomo_queue()
obj.write_pdf_report = lambda: None
obj.sub_tomo_scan = lambda subtomo_number, n_subtomos=2, start_angle=None: None
obj.omnygui_show_progress = lambda: None
monkeypatch.setitem(
builtins.__dict__,
"bec",
types.SimpleNamespace(
active_account=active_account, queue=types.SimpleNamespace(next_scan_number=1)
),
obj.OMNYTools = types.SimpleNamespace(printgreenbold=lambda msg: None)
# write_to_scilog() reads the module-level `bec` (captured at import
# time), not builtins["bec"] -- both must point at the same fake so
# scilog._enabled resolves to False (no scilog configured) rather than
# AttributeError-ing on a stale/missing bec.messaging.
fake_bec = types.SimpleNamespace(
active_account=active_account,
queue=types.SimpleNamespace(next_scan_number=1),
messaging=types.SimpleNamespace(scilog=types.SimpleNamespace(_enabled=False)),
)
monkeypatch.setitem(builtins.__dict__, "bec", fake_bec)
monkeypatch.setattr(omny_module, "bec", fake_bec, raising=False)
monkeypatch.setitem(builtins.__dict__, "scans", _FakeScans())
return obj
@@ -195,6 +205,74 @@ def test_tomo_scan_sets_timing_fields_on_new_scan(monkeypatch):
assert omny.progress["accumulated_idle_time"] == 0.0
# --- idle-time tracking + end-of-scan measurement-time summary --------------
#
# Mirrors flomni's identical _tomo_scan_at_angle() heartbeat-gap heuristic
# and tomo_scan()'s end-of-scan timing summary -- omny previously tracked
# neither (heartbeat was only ever cleared to None, never refreshed during
# a running scan, so accumulated_idle_time stayed 0 forever and no summary
# was ever printed).
def test_tomo_scan_at_angle_tracks_idle_time_from_heartbeat_gap(monkeypatch, capsys):
omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
omny.tomo_countingtime = 0.1 # normal_cadence = max(60, 0.5) = 60s
old_heartbeat = datetime.datetime.now() - datetime.timedelta(seconds=200)
omny.progress["heartbeat"] = old_heartbeat.isoformat()
omny.progress["accumulated_idle_time"] = 0.0
omny._print_progress = lambda: None
omny.special_angles = []
omny._current_special_angles = []
omny._start_beam_check = lambda: None
omny._was_beam_okay = lambda: True
omny._at_each_angle = lambda angle: None
omny._write_tomo_scan_number = lambda *a: None
omny._tomo_scan_at_angle(10.0, 1)
idle = omny.progress["accumulated_idle_time"]
assert idle > 100 # ~200s gap - 60s normal cadence
assert "Detected a" in capsys.readouterr().out
# heartbeat refreshed to "now" so the busy-detector sees a running scan
assert omny.progress["heartbeat"] != old_heartbeat.isoformat()
def test_tomo_scan_at_angle_no_idle_time_within_normal_cadence(monkeypatch, capsys):
omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
omny.tomo_countingtime = 0.1
recent_heartbeat = datetime.datetime.now() - datetime.timedelta(seconds=5)
omny.progress["heartbeat"] = recent_heartbeat.isoformat()
omny.progress["accumulated_idle_time"] = 0.0
omny._print_progress = lambda: None
omny.special_angles = []
omny._current_special_angles = []
omny._start_beam_check = lambda: None
omny._was_beam_okay = lambda: True
omny._at_each_angle = lambda angle: None
omny._write_tomo_scan_number = lambda *a: None
omny._tomo_scan_at_angle(10.0, 1)
assert omny.progress["accumulated_idle_time"] == 0.0
assert "Detected a" not in capsys.readouterr().out
def test_tomo_scan_prints_measurement_time_summary(monkeypatch, capsys):
omny = make_omny_for_tomo_scan(monkeypatch, 45.0, active_account="")
def _fake_sub_tomo_scan(subtomo_number, n_subtomos=2, start_angle=None):
omny.progress["accumulated_idle_time"] = 10.0
omny.sub_tomo_scan = _fake_sub_tomo_scan
omny.tomo_scan()
out = capsys.readouterr().out
assert "Total measurement time:" in out
assert "Total measurement time excluding detected gaps:" in out
assert "Total measurement time lost to detected gaps:" in out
# --- write_to_scilog(): checks scilog._enabled before attempting a send -----
#
# The old bec.logbook/client.logbook.send_logbook_message() API had no such