Scan errors go to the session log and the scan info; resume_tries resumes a failed step automatically
A step that raised was printed and swallowed by scan_all, so no command failed and neither eco.logs nor %ei ever saw it. Every failed step is now logged as an error (kernel_registry.log_caught_error, with scan, step, traceback and resume command; eco.logs shows the traceback's last lines) and kept in scan_info["errors"], with scan_info["outcome"] saying how the scan ended. A failed step is taken back out (scan info, and the counters' new callbacks_step_failed), which also fixes the duplicate step a manual resume left. resume_tries=n redoes it up to n times; default 1 in a queue, 0 elsewhere.
This commit is contained in:
1 parent
d0a1518519
commit
c7e51b285b
7 files changed
+295
-18
No files matched your search
@@ -67,6 +67,13 @@ Every queue method is also a `%queue` sub-command:
|
||||
When a task fails, the queue pauses so later tasks don't run on a broken
|
||||
step. Look at the error, then `%queue resume`.
|
||||
|
||||
A scan is different: it catches an error in one of its steps itself. In a
|
||||
queue it redoes the failed step once (`resume_tries`, default 1 there, see
|
||||
{doc}`scanning`); if that fails too, the scan ends early but the task counts
|
||||
as done and the queue goes on with the next one. The error is in the session
|
||||
log (`%ei`) and in the scan's `scan_info["errors"]` either way, and the scan
|
||||
prints the command that resumes it.
|
||||
|
||||
## Several queues
|
||||
|
||||
```python
|
||||
|
||||
@@ -220,6 +220,27 @@ same scan: the counters' monitors and live plots carry on, and the data of
|
||||
all steps end up together. If the scan moved back to its start values when it
|
||||
ended, the next step moves to where it left off first.
|
||||
|
||||
**Resuming automatically.** `resume_tries=n` lets a scan get over a passing
|
||||
error by itself: a step that raises is taken back out and done again, up to
|
||||
`n` times per scan, before the scan ends as above:
|
||||
|
||||
```python
|
||||
scans.ascan(mot, 0, 1, 40, 100, resume_tries=2)
|
||||
```
|
||||
|
||||
Inside a queue the default is 1 (one retry, so an unattended scan survives a
|
||||
glitch and the queue goes on); elsewhere 0. Ctrl-C is never resumed.
|
||||
|
||||
**Every failed step is recorded**, also one that was resumed:
|
||||
|
||||
- as an error in the session log, with the scan, the step, the traceback and
|
||||
the resume command -- so `eco.logs` and `%ei` find it although the scan
|
||||
caught it and no command failed;
|
||||
- in the scan's info, `scan.scan_info["errors"]` (type, message, traceback,
|
||||
step, whether resumed), next to `scan.scan_info["outcome"]` (`complete`,
|
||||
`aborted`, `stopped` or `interrupted`, steps done, automatic resumes);
|
||||
with the DAQ both are in the run's scan info file.
|
||||
|
||||
```{admonition} Did you know?
|
||||
:class: tip
|
||||
|
||||
|
||||
@@ -73,6 +73,8 @@ class CounterValue:
|
||||
# a scan resumed after it ended unfinished (StepScan.scan_all again):
|
||||
# the end callbacks above stopped the monitors and the plot
|
||||
self.callbacks_resume_scan = [self.resume_monitoring, self.resume_animation]
|
||||
# a step that raised is redone: drop what was recorded for it
|
||||
self.callbacks_step_failed = [self.forget_failed_step]
|
||||
self.name = name
|
||||
|
||||
@mainthread.on_main # the animation's timer lives on the main thread
|
||||
@@ -121,6 +123,15 @@ class CounterValue:
|
||||
if hasattr(state, attribute):
|
||||
delattr(state, attribute)
|
||||
|
||||
def forget_failed_step(self, scan=None, **kwargs):
|
||||
"""Keep only the detector values and acquisition windows of the
|
||||
steps the scan counts as done (``scan.next_step``)."""
|
||||
state = self._state(scan)
|
||||
for attribute in ("timestamp_intervals", "detector_values"):
|
||||
recorded = getattr(state, attribute, None)
|
||||
if recorded is not None and len(recorded) > scan.next_step:
|
||||
del recorded[scan.next_step :]
|
||||
|
||||
def resume_monitoring(self, scan=None, **kwargs):
|
||||
"""Restart the monitors stopped at the end of the scan; they keep
|
||||
what they collected and add to it."""
|
||||
|
||||
+158
-18
@@ -121,6 +121,10 @@ class StepScan(Assembly):
|
||||
counters, kwargs_callbacks.pop("counter_options", None) or []
|
||||
)
|
||||
self.step_window = {} # see set_step_window
|
||||
# how often a failed step is retried by resuming automatically
|
||||
# (None: 1 in a queue, else 0 -- see effective_resume_tries)
|
||||
self.resume_tries = kwargs_callbacks.pop("resume_tries", None)
|
||||
self._auto_resumes = 0
|
||||
|
||||
self._append(
|
||||
DetectorGet, lambda: self._get_names(self.counters), name="counters_names"
|
||||
@@ -379,6 +383,108 @@ class StepScan(Assembly):
|
||||
print(f"{name}: could not resume ({type(exc).__name__}: {exc})")
|
||||
self._have_run_callbacks_end_scan = False
|
||||
|
||||
def effective_resume_tries(self):
|
||||
"""`resume_tries` as set, else 1 while running in a queue (an
|
||||
unattended scan should get over a passing error) and 0 otherwise."""
|
||||
if self.resume_tries is not None:
|
||||
return int(self.resume_tries)
|
||||
return 1 if task_queue.in_queue() else 0
|
||||
|
||||
def _rollback_failed_step(self):
|
||||
"""Take a step that raised back out, so redoing it (a resume) does
|
||||
not leave a second entry: the step's scan info, and what counters
|
||||
recorded for it (their ``callbacks_step_failed``)."""
|
||||
info = self.scan_info.get("scan_step_info", [])
|
||||
if len(info) > self.next_step:
|
||||
try:
|
||||
self.remove_last_scan_info_entry()
|
||||
except Exception:
|
||||
pass
|
||||
for ctr in self.counters:
|
||||
for tcb in getattr(ctr, "callbacks_step_failed", None) or []:
|
||||
try:
|
||||
tcb(self, **self.callbacks_kwargs)
|
||||
except Exception as exc:
|
||||
name = getattr(ctr, "name", None) or repr(ctr)
|
||||
print(f"{name}: could not drop the failed step ({type(exc).__name__}: {exc})")
|
||||
|
||||
def _scan_name(self):
|
||||
command = self.resume_command()
|
||||
if command:
|
||||
return command[: -len(".scan_all()")]
|
||||
try:
|
||||
return self.alias.get_full_name()
|
||||
except Exception:
|
||||
return getattr(self, "name", "scan")
|
||||
|
||||
def _record_scan_error(self, exc, tb, resumed):
|
||||
"""A failed step, kept where it can be found later: in the scan's
|
||||
info (``scan_info["errors"]``, written to its file when it has one)
|
||||
and as an error in the session log (eco.logs, %ei) -- the scan
|
||||
catches the error, so no failed command would show it there."""
|
||||
try:
|
||||
total = self.expected_number_of_steps()
|
||||
except Exception:
|
||||
total = None
|
||||
entry = {
|
||||
"time": time(),
|
||||
"step": self.next_step,
|
||||
"type": type(exc).__name__,
|
||||
"message": str(exc),
|
||||
"traceback": tb,
|
||||
"resumed": resumed,
|
||||
}
|
||||
self.scan_info.setdefault("errors", []).append(entry)
|
||||
what = "resumed automatically" if resumed else "scan aborted"
|
||||
where = f"step {self.next_step + 1}" + (f" of {total}" if total else "")
|
||||
try:
|
||||
from eco.widgets.kernel_registry import log_caught_error
|
||||
|
||||
log_caught_error(
|
||||
type(exc).__name__,
|
||||
f"{exc} [scan {self._scan_name()}, {where}, {what}]",
|
||||
traceback=tb,
|
||||
scan=self._scan_name(),
|
||||
step=self.next_step,
|
||||
resumed=resumed,
|
||||
resume_command=self.resume_command(),
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
self._write_scan_info_quietly()
|
||||
|
||||
def _record_outcome(self, done, interrupted=False):
|
||||
"""How the scan ended, in ``scan_info["outcome"]``: "complete",
|
||||
"stopped" (request_stop), "interrupted" (Ctrl-C) or "aborted" (an
|
||||
error), with the steps done and the automatic resumes."""
|
||||
if done:
|
||||
state = "complete"
|
||||
elif interrupted:
|
||||
state = "interrupted"
|
||||
elif self._stop_requested:
|
||||
state = "stopped"
|
||||
else:
|
||||
state = "aborted"
|
||||
try:
|
||||
total = self.expected_number_of_steps()
|
||||
except Exception:
|
||||
total = None
|
||||
self.scan_info["outcome"] = {
|
||||
"state": state,
|
||||
"steps_done": self.next_step,
|
||||
"steps_expected": total,
|
||||
"automatic_resumes": self._auto_resumes,
|
||||
"time": time(),
|
||||
}
|
||||
self._write_scan_info_quietly()
|
||||
|
||||
def _write_scan_info_quietly(self):
|
||||
if getattr(self, "scan_info_filename", None):
|
||||
try:
|
||||
self.writeScanInfo()
|
||||
except Exception as exc:
|
||||
print(f"could not write the scan info ({type(exc).__name__}: {exc})")
|
||||
|
||||
def resume_command(self):
|
||||
"""The command that resumes this scan, as typed at the session's
|
||||
prompt (e.g. ``scans.acquiring_scan.scan_all()``), or None if the
|
||||
@@ -832,9 +938,11 @@ class StepScan(Assembly):
|
||||
elif self._have_run_callbacks_end_scan:
|
||||
# resuming a scan that ended before its last step
|
||||
self._stop_requested = False
|
||||
self._auto_resumes = 0
|
||||
self.run_callbacks_resume_scan()
|
||||
task_queue.register_scan(self)
|
||||
done = False
|
||||
interrupted = False
|
||||
first_step = self.next_step # > 0 when resuming
|
||||
# no live progress bar from a queue's worker thread: it would redraw
|
||||
# over the prompt the user is typing at
|
||||
@@ -844,31 +952,20 @@ class StepScan(Assembly):
|
||||
total=self.expected_number_of_steps() - first_step,
|
||||
)
|
||||
try:
|
||||
while not done:
|
||||
# sliced while paused, so open Qt windows stay alive (see
|
||||
# eco.utilities.mainthread.wait_event)
|
||||
mainthread.wait_event(self._pause_event)
|
||||
if self._stop_requested:
|
||||
break
|
||||
done = not self.do_next_step(step_info=step_info)
|
||||
# total is re-read every step: steps can be added while
|
||||
# the scan runs (adaptive scans), and a repeated step
|
||||
# does not count as progress
|
||||
self._progress.update(
|
||||
pr_task,
|
||||
completed=self.next_step - first_step,
|
||||
total=self.expected_number_of_steps() - first_step,
|
||||
)
|
||||
except:
|
||||
done, tb = self._run_steps(step_info, pr_task, first_step)
|
||||
except BaseException: # Ctrl-C (an error in a step is handled in _run_steps)
|
||||
interrupted = True
|
||||
tb = traceback.format_exc()
|
||||
else:
|
||||
tb = "Ended all steps without interruption."
|
||||
self._rollback_failed_step() # it may have hit in the middle of a step
|
||||
finally:
|
||||
self._progress.stop()
|
||||
print(tb)
|
||||
if self.timing.n_steps:
|
||||
print(self.timing.summary())
|
||||
|
||||
# before the end callbacks: the DAQ's copy of the scan info
|
||||
# then has it too
|
||||
self._record_outcome(done, interrupted=interrupted)
|
||||
self.run_callbacks_end_scan()
|
||||
|
||||
if self.return_at_end == "question":
|
||||
@@ -914,6 +1011,49 @@ class StepScan(Assembly):
|
||||
if not done:
|
||||
self._report_unfinished()
|
||||
|
||||
def _run_steps(self, step_info, pr_task, first_step):
|
||||
"""The steps until the last one, a stop request, or an error that
|
||||
is not resumed: (done, text to print). A step that raises is taken
|
||||
back out and recorded (_record_scan_error); while
|
||||
effective_resume_tries() allows, the scan goes on with that step
|
||||
again, otherwise it ends there. Ctrl-C is not caught here."""
|
||||
while True:
|
||||
# sliced while paused, so open Qt windows stay alive (see
|
||||
# eco.utilities.mainthread.wait_event)
|
||||
mainthread.wait_event(self._pause_event)
|
||||
if self._stop_requested:
|
||||
return False, "Stopped before the last step."
|
||||
try:
|
||||
more = self.do_next_step(step_info=step_info)
|
||||
except Exception as exc:
|
||||
tb = traceback.format_exc()
|
||||
self._rollback_failed_step()
|
||||
resume = (
|
||||
self._auto_resumes < self.effective_resume_tries()
|
||||
and not self._stop_requested
|
||||
)
|
||||
self._record_scan_error(exc, tb, resumed=resume)
|
||||
if not resume:
|
||||
return False, tb
|
||||
self._auto_resumes += 1
|
||||
print(tb)
|
||||
print(
|
||||
f"{colorama.Style.BRIGHT}{colorama.Fore.YELLOW}Step {self.next_step + 1} failed "
|
||||
f"({type(exc).__name__}: {exc}); resuming automatically "
|
||||
f"({self._auto_resumes} of {self.effective_resume_tries()}){colorama.Style.RESET_ALL}"
|
||||
)
|
||||
continue
|
||||
if not more:
|
||||
return True, "Ended all steps without interruption."
|
||||
# total is re-read every step: steps can be added while the
|
||||
# scan runs (adaptive scans), and a repeated step does not
|
||||
# count as progress
|
||||
self._progress.update(
|
||||
pr_task,
|
||||
completed=self.next_step - first_step,
|
||||
total=self.expected_number_of_steps() - first_step,
|
||||
)
|
||||
|
||||
def changeToInitialValues(self):
|
||||
c = []
|
||||
for adj, iv in zip(self.adjustables, self.initial_values()):
|
||||
|
||||
@@ -196,6 +196,7 @@ def frequent_commands(n=200, max_scan=5000):
|
||||
#: (hundreds of session files, 100k+ lines) are what made the viewer take
|
||||
#: many seconds to open; pass days=None for everything.
|
||||
DEFAULT_DAYS = 30
|
||||
TRACEBACK_LINES = 12 # of a caught error's traceback shown with it (log_caught_error)
|
||||
|
||||
|
||||
def kernel_loader(session=None, days=DEFAULT_DAYS):
|
||||
@@ -345,6 +346,10 @@ def _kernel_record_to_entry(rec, session_label):
|
||||
kind, text = "widget_control", rec.get("code", "")
|
||||
elif event == "error":
|
||||
kind, text = "error", f"{rec.get('ename', '')}: {rec.get('evalue', '')}"
|
||||
if rec.get("traceback"):
|
||||
# caught errors (kernel_registry.log_caught_error): where it happened
|
||||
lines = [l for l in str(rec["traceback"]).rstrip().splitlines() if l.strip()]
|
||||
text += "\n" + "\n".join(lines[-TRACEBACK_LINES:])
|
||||
elif event == "warning":
|
||||
# see kernel_registry.install_warning_capture
|
||||
kind, text = "warning", f"{rec.get('category', 'Warning')}: {rec.get('text', '')}"
|
||||
|
||||
@@ -212,6 +212,30 @@ def unregister(session):
|
||||
_registry.remove(session)
|
||||
|
||||
|
||||
def log_caught_error(ename, evalue, traceback=None, **fields):
|
||||
"""Record an error that eco caught and reported itself (a scan step
|
||||
that failed) as an "error" event in this session's log, where a failed
|
||||
command would have put it -- so eco.logs and %ei find it although no
|
||||
command failed. `traceback`: the formatted traceback (eco.logs shows its
|
||||
last lines with the error). The terminal session's logger is used, else
|
||||
the first registered session of this process; nothing is raised."""
|
||||
try:
|
||||
from IPython import get_ipython
|
||||
|
||||
session = getattr(get_ipython(), "_eco_shell_logger_session", None)
|
||||
if session is None:
|
||||
sessions = all_sessions()
|
||||
session = sessions[0] if sessions else None
|
||||
if session is None:
|
||||
return False
|
||||
if traceback:
|
||||
fields["traceback"] = traceback
|
||||
session.log_output("error", ename=ename, evalue=evalue, caught=True, **fields)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def all_sessions():
|
||||
with _registry_lock:
|
||||
return list(_registry)
|
||||
|
||||
@@ -101,3 +101,72 @@ def test_a_detectors_own_scan_is_resumed_through_the_detector(sim, monkeypatch,
|
||||
finally:
|
||||
detector.scans.callbacks_end_step.pop()
|
||||
assert "> sim.peak.signal.scans.acquiring_scan.scan_all()" in capsys.readouterr().out
|
||||
|
||||
|
||||
# -- failed steps: taken back out, recorded, resumed automatically ---------------
|
||||
|
||||
|
||||
def test_a_resumed_scan_has_each_step_once(sim, monkeypatch):
|
||||
scans, s = aborted_scan(sim, monkeypatch, namespace=lambda scans: {"scans": scans})
|
||||
assert len(s.scan_info["scan_step_info"]) == len(s.timestamp_intervals) == 2
|
||||
s.scan_all()
|
||||
assert s.next_step == len(s.scan_info["scan_step_info"]) == len(s.timestamp_intervals) == 5
|
||||
|
||||
|
||||
def test_an_aborted_scan_records_its_error_and_outcome(sim, monkeypatch):
|
||||
logged = []
|
||||
monkeypatch.setattr(
|
||||
"eco.widgets.kernel_registry.log_caught_error",
|
||||
lambda ename, evalue, **fields: logged.append((ename, evalue, fields)),
|
||||
)
|
||||
scans, s = aborted_scan(sim, monkeypatch, namespace=lambda scans: {"scans": scans})
|
||||
(error,) = s.scan_info["errors"]
|
||||
assert error["type"] == "RuntimeError" and error["step"] == 2 and not error["resumed"]
|
||||
assert "lost the beam" in error["traceback"]
|
||||
assert s.scan_info["outcome"]["state"] == "aborted"
|
||||
assert s.scan_info["outcome"]["steps_done"] == 2
|
||||
((ename, evalue, fields),) = logged
|
||||
assert ename == "RuntimeError"
|
||||
assert "lost the beam [scan scans.acquiring_scan, step 3 of 5, scan aborted]" == evalue
|
||||
assert fields["resume_command"] == "scans.acquiring_scan.scan_all()"
|
||||
s.scan_all()
|
||||
assert s.scan_info["outcome"]["state"] == "complete"
|
||||
|
||||
|
||||
def test_resume_tries_gets_over_a_failed_step(sim, monkeypatch, capsys):
|
||||
monkeypatch.setattr(scan_module, "get_ipython", lambda: None)
|
||||
scans = Scans(callbacks_end_step=[failing_once(2)])
|
||||
s = scans.ascan(sim.peak.x, 0, 1, 4, 0.05, counters=[sim.peak.signal],
|
||||
return_at_end=False, resume_tries=1)
|
||||
out = capsys.readouterr().out
|
||||
assert "resuming automatically (1 of 1)" in out and "Scan aborted" not in out
|
||||
assert s.next_step == len(s.scan_info["scan_step_info"]) == len(s.timestamp_intervals) == 5
|
||||
assert s.scan_info["outcome"] == dict(s.scan_info["outcome"], state="complete", automatic_resumes=1)
|
||||
assert [e["resumed"] for e in s.scan_info["errors"]] == [True]
|
||||
|
||||
|
||||
def test_in_a_queue_a_scan_resumes_once_by_default(sim, monkeypatch):
|
||||
s = scan_module.StepScan([sim.peak.x], [[0.0]], counters=[sim.peak.signal], Npulses=0.05)
|
||||
assert s.effective_resume_tries() == 0
|
||||
monkeypatch.setattr(scan_module.task_queue, "in_queue", lambda: True)
|
||||
assert s.effective_resume_tries() == 1
|
||||
s.resume_tries = 3
|
||||
assert s.effective_resume_tries() == 3
|
||||
|
||||
|
||||
def test_a_caught_error_goes_to_the_session_log_with_its_traceback(monkeypatch):
|
||||
from eco import logs
|
||||
from eco.widgets import kernel_registry
|
||||
|
||||
records = []
|
||||
|
||||
class Session:
|
||||
def log_output(self, event, **fields):
|
||||
records.append(dict(fields, event=event, t=1.0))
|
||||
|
||||
monkeypatch.setattr("IPython.get_ipython", lambda: SimpleNamespace(_eco_shell_logger_session=Session()))
|
||||
tb = "Traceback (most recent call last):\n File \"x.py\", line 3, in step\nRuntimeError: lost the beam\n"
|
||||
assert kernel_registry.log_caught_error("RuntimeError", "lost the beam [scan s]", traceback=tb, step=2)
|
||||
entry = logs._kernel_record_to_entry(records[0], "console")
|
||||
assert entry.is_error and entry.text.startswith("RuntimeError: lost the beam [scan s]")
|
||||
assert 'File "x.py", line 3, in step' in entry.text
|
||||
Reference in new issue
Block a user