diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py index 0b9a1c82..93f178ba 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py @@ -2288,7 +2288,7 @@ class LamNI( ) self.write_to_scilog(timing_content, ["tomoscan"]) - def tomo_scan_resume(self) -> None: + def tomo_scan_resume(self, from_queue: bool = False) -> None: """Resume a tomo_scan() that crashed or was interrupted, picking up automatically from wherever ``progress`` last reported -- no need to read the last subtomo/angle/projection off the progress printout by @@ -2305,11 +2305,20 @@ class LamNI( internally when resuming a job that previously failed partway through, so a failed queue job picks up mid-scan rather than restarting from subtomo 1 / angle 0. + + If the tomo queue holds an unfinished ("incomplete"/"running") job, + a manual call warns and asks for confirmation, since resuming by + hand bypasses the queue (use tomo_queue_resume() instead). + ``from_queue`` is set by tomo_queue_execute() to skip that prompt. """ if self.progress.get("tomo_start_time") is None: print("No tomo scan in progress to resume -- nothing to do.") return + if not from_queue and not self._confirm_resume_outside_queue(): + print("Aborted.") + return + if self.tomo_type == 1: subtomo_start = self.progress["subtomo"] start_angle = self.progress["angle"] diff --git a/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py b/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py index b43f1cab..efa0a2c9 100644 --- a/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py +++ b/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/tomo_queue_mixin.py @@ -934,7 +934,9 @@ class TomoQueueMixin: for name, value in job["params"].items(): setattr(self, name, value) if resume_job: - self.tomo_scan_resume() + # from_queue=True: skip tomo_scan_resume()'s + # "unfinished queue job" prompt -- this *is* the queue. + self.tomo_scan_resume(from_queue=True) else: # interactive=False: an unattended queued run must # never block on input() -- see LamNI.tomo_scan()'s @@ -1005,6 +1007,35 @@ class TomoQueueMixin: "tomo_queue_reacquire() on a tomo_type-1 job." ) + def _confirm_resume_outside_queue(self) -> bool: + """Guard for a manual ``tomo_scan_resume()`` call. + + If the queue holds an "incomplete"/"running" job (e.g. after a + Ctrl-C abort, which leaves it "running"), resuming by hand bypasses + the queue: the job stays unfinished there, so a later + ``tomo_queue_resume()`` would resume it again and the jobs after it + would not run. Warn, print the hint, and ask. Returns True if the + resume should go ahead (nothing unfinished in the queue, or the + operator confirmed). + """ + unfinished = [ + j for j in self._tomo_queue_proxy.as_list() if j["status"] in ("incomplete", "running") + ] + if not unfinished: + return True + print("\nWARNING: the tomo queue has an unfinished job:") + for job in unfinished: + print( + f" {job['status']:>10s} {job['label']:<20s} {self._describe_tomo_job_line(job)}" + ) + print( + "Use tomo_queue_resume() to resume it and continue with the rest of the queue.\n" + "If you resume manually instead, the queue is not told -- mark the job done " + "afterwards, or it will be resumed again:\n" + " self._tomo_queue_proxy.update_by_id(, status='done')" + ) + return self.OMNYTools.yesno("Run tomo_scan_resume() anyway, bypassing the queue?", "n") + def tomo_queue_resume(self, start_index: int = 0) -> None: """Continue running the tomo queue from wherever it currently stands -- an alias for ``tomo_queue_execute()``, added purely so diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py index 19ae3446..3a8469d6 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py @@ -3050,7 +3050,7 @@ class Flomni( ) self._scilog_write(timing_content, "tomoscan") - def tomo_scan_resume(self) -> None: + def tomo_scan_resume(self, from_queue: bool = False) -> None: """Resume a tomo_scan() that crashed or was interrupted, picking up automatically from wherever ``progress`` last reported -- no need to read the last subtomo/angle/projection off the progress printout by @@ -3071,11 +3071,20 @@ class Flomni( an explicit abort (e.g. Ctrl-C) -- it is not needed just to pick up angle lines appended to the file while a scan is running, since tomo_scan() already waits for and picks those up on its own. + + If the tomo queue holds an unfinished ("incomplete"/"running") job, + a manual call warns and asks for confirmation, since resuming by + hand bypasses the queue (use tomo_queue_resume() instead). + ``from_queue`` is set by tomo_queue_execute() to skip that prompt. """ if self.progress.get("tomo_start_time") is None: print("No tomo scan in progress to resume -- nothing to do.") return + if not from_queue and not self._confirm_resume_outside_queue(): + print("Aborted.") + return + if self.tomo_type == 1: subtomo_start = self.progress["subtomo"] start_angle = self.progress["angle"] diff --git a/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/TODO_tomo_scan_resume_queue_guard.md b/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/TODO_tomo_scan_resume_queue_guard.md new file mode 100644 index 00000000..ea20d79d --- /dev/null +++ b/csaxs_bec/bec_ipython_client/plugins/omny/AI_docs/TODO_tomo_scan_resume_queue_guard.md @@ -0,0 +1,25 @@ +# TODO: port the "unfinished queue job" guard on `tomo_scan_resume()` to OMNY + +**Status: implemented on Flomni and LamNI (2026-10), not applicable to OMNY yet** -- +OMNY has no `TomoQueueMixin`, no tomo queue and no `tomo_scan_resume()`. + +## What landed on Flomni/LamNI +- Problem: a Ctrl-C abort of `tomo_queue_execute()` leaves the job "running" (only + `Exception` is caught -> "incomplete"). Calling `tomo_scan_resume()` by hand then finishes + the scan but the queue still has the job unfinished, so a later `tomo_queue_resume()` + resumes it again (at the last projection) and the jobs after it never ran. +- `TomoQueueMixin._confirm_resume_outside_queue()` + (`OMNY_shared/tomo_queue_mixin.py`): if the queue has an "incomplete"/"running" job, print a + warning + hint (use `tomo_queue_resume()`, or mark the job done via + `_tomo_queue_proxy.update_by_id(id, status="done")`) and ask yes/no (default no). +- `tomo_scan_resume(from_queue=False)` in `flomni/flomni.py` and `LamNI/lamni.py` calls it + after the "nothing to resume" check; `tomo_queue_execute()` passes `from_queue=True` so the + unattended queue never prompts. + +## To do for OMNY +When the tomo queue (`TomoQueueMixin`) and a `tomo_scan_resume()` are ported to OMNY, give +`tomo_scan_resume()` the same `from_queue` parameter and guard call, and make sure +`tomo_queue_execute()` calls it with `from_queue=True`. + +Related, not done anywhere: catching `KeyboardInterrupt` in `tomo_queue_execute()` so an abort +marks the job "incomplete" instead of leaving it "running". diff --git a/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py b/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py index 0397b2d8..fdc43c3e 100644 --- a/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py +++ b/tests/tests_bec_ipython_client/test_lamni_tomo_queue.py @@ -11,7 +11,7 @@ import types import csaxs_bec.bec_ipython_client.plugins.LamNI.lamni as lamni_module import csaxs_bec.bec_ipython_client.plugins.OMNY_shared.tomo_queue_mixin as tomo_queue_mixin -from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI, LamNIError +from csaxs_bec.bec_ipython_client.plugins.LamNI.lamni import LamNI, LamNIError, _ProgressProxy class _FakeRtx: @@ -159,7 +159,7 @@ def test_tomo_queue_execute_resumes_incomplete_job(): lamni.tomo_queue_execute() - assert calls == [("resume", (), {})] + assert calls == [("resume", (), {"from_queue": True})] job = lamni._tomo_queue_proxy.as_list()[0] assert job["status"] == "done" @@ -302,3 +302,44 @@ def test_at_each_angle_hook_refreshed_after_module_reload(tmp_path, capsys): finally: sys.path.remove(str(tmp_path)) sys.modules.pop(module_name, None) + + +def _resumable_lamni(status=None, answer=True): + lamni = make_lamni() + calls = [] + lamni.tomo_scan = lambda *a, **k: calls.append(("scan", a, k)) + lamni.tomo_type = 2 + lamni._progress_proxy = _ProgressProxy(lamni.client) + lamni.progress["tomo_start_time"] = "2026-01-01T00:00:00" + lamni.progress["projection"] = 5 + lamni.OMNYTools = types.SimpleNamespace(yesno=lambda *a, **k: answer) + if status: + index = lamni.tomo_queue_add(label="job1") + lamni._tomo_queue_proxy.update(index, status=status) + return lamni, calls + + +def test_tomo_scan_resume_no_unfinished_job_does_not_prompt(): + lamni, calls = _resumable_lamni(answer=False) # would abort if prompted + lamni.tomo_scan_resume() + assert calls == [("scan", (), {"projection_number": 5})] + + +def test_tomo_scan_resume_warns_and_aborts_on_no(capsys): + lamni, calls = _resumable_lamni(status="running", answer=False) + lamni.tomo_scan_resume() + assert calls == [] + out = capsys.readouterr().out + assert "unfinished job" in out and "tomo_queue_resume()" in out + + +def test_tomo_scan_resume_proceeds_on_yes(): + lamni, calls = _resumable_lamni(status="incomplete", answer=True) + lamni.tomo_scan_resume() + assert calls == [("scan", (), {"projection_number": 5})] + + +def test_tomo_scan_resume_from_queue_never_prompts(): + lamni, calls = _resumable_lamni(status="running", answer=False) + lamni.tomo_scan_resume(from_queue=True) + assert calls == [("scan", (), {"projection_number": 5})]