diff --git a/docs/queues.md b/docs/queues.md index 099b7b3..9e68935 100644 --- a/docs/queues.md +++ b/docs/queues.md @@ -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 diff --git a/docs/scanning.md b/docs/scanning.md index f4eda3d..ab7bcb6 100644 --- a/docs/scanning.md +++ b/docs/scanning.md @@ -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 diff --git a/eco/acquisition/counters.py b/eco/acquisition/counters.py index 7dbe9db..3d7c5fe 100644 --- a/eco/acquisition/counters.py +++ b/eco/acquisition/counters.py @@ -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.""" diff --git a/eco/acquisition/scan.py b/eco/acquisition/scan.py index 92997df..334a14a 100755 --- a/eco/acquisition/scan.py +++ b/eco/acquisition/scan.py @@ -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()): diff --git a/eco/logs.py b/eco/logs.py index 5ebddc8..0e152a2 100644 --- a/eco/logs.py +++ b/eco/logs.py @@ -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', '')}" diff --git a/eco/widgets/kernel_registry.py b/eco/widgets/kernel_registry.py index 521f75a..12b756d 100644 --- a/eco/widgets/kernel_registry.py +++ b/eco/widgets/kernel_registry.py @@ -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) diff --git a/tests/test_scan_resume.py b/tests/test_scan_resume.py index 6ba444c..59c8ad2 100644 --- a/tests/test_scan_resume.py +++ b/tests/test_scan_resume.py @@ -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