Feature/custom exit codes in tools

docs/changes/2566.feature.rst

@@ -0,0 +1,7 @@
+Add ``SystemExit`` handling at the ``ctapipe.core.Tool`` level
+
+If a ``SystemExit`` with a custom error code is generated during the tool execution,
+the tool will be terminated gracefully and the error code will be preserved and propagated.
+
+The ``Activity`` statuses have been updated to ``["running", "success", "interrupted", "error"]``.
+The ``"running"`` status is assigned at init.

src/ctapipe/core/provenance.py

@@ -4,6 +4,7 @@
 TODO: have this register whenever ctapipe is loaded
 
 """
+
 import json
 import logging
 import os
@@ -145,7 +146,7 @@ def add_config(self, config):
             activity.name,
         )
 
-    def finish_activity(self, status="completed", activity_name=None):
+    def finish_activity(self, status="completed", exit_code=0, activity_name=None):
         """end the current activity"""
         activity = self._activities.pop()
         if activity_name is not None and activity_name != activity.name:
@@ -155,7 +156,7 @@ def finish_activity(self, status="completed", activity_name=None):
                 )
             )
 
-        activity.finish(status)
+        activity.finish(status, exit_code)
         self._finished_activities.append(activity)
         log.debug(f"finished activity: {activity.name}")
 
@@ -221,11 +222,13 @@ def __init__(self, activity_name=sys.executable):
         self._prov = {
             "activity_name": activity_name,
             "activity_uuid": str(uuid.uuid4()),
+            "status": "running",
             "start": {},
             "stop": {},
             "system": {},
             "input": [],
             "output": [],
+            "exit_code": None,
         }
         self.name = activity_name
 
@@ -268,14 +271,15 @@ def register_config(self, config):
         """add a dictionary of configuration parameters to this activity"""
         self._prov["config"] = config
 
-    def finish(self, status="completed"):
+    def finish(self, status="success", exit_code=0):
         """record final provenance information, normally called at shutdown."""
         self._prov["stop"].update(_sample_cpu_and_memory())
 
         # record the duration (wall-clock) for this activity
         t_start = Time(self._prov["start"]["time_utc"], format="isot")
         t_stop = Time(self._prov["stop"]["time_utc"], format="isot")
         self._prov["status"] = status
+        self._prov["exit_code"] = exit_code
         self._prov["duration_min"] = (t_stop - t_start).to("min").value
 
     @property

src/ctapipe/core/tests/test_run_tool.py

@@ -1,3 +1,4 @@
+import sys
 from subprocess import CalledProcessError
 
 import pytest
@@ -15,5 +16,16 @@ def start(self):
 
     ret = run_tool(ErrorTool(), ["--non-existing-alias"], raises=False)
     assert ret == 2
+
+    class SysExitTool(Tool):
+        def setup(self):
+            pass
+
+        def start(self):
+            sys.exit(4)
+
+    ret = run_tool(SysExitTool(), raises=False)
+    assert ret == 4
+
     with pytest.raises(CalledProcessError):
         run_tool(ErrorTool(), ["--non-existing-alias"], raises=True)

src/ctapipe/core/tool.py

@@ -1,4 +1,5 @@
 """Classes to handle configurable command-line user interfaces."""
+
 import html
 import logging
 import logging.config
@@ -255,7 +256,7 @@ def initialize(self, argv=None):
         self.update_config(self.cli_config)
         self.update_logging_config()
 
-        self.log.info(f"ctapipe version {self.version_string}")
+        self.log.info("ctapipe version %s", self.version_string)
 
     def load_config_file(self, path: str | pathlib.Path) -> None:
         """
@@ -405,15 +406,15 @@ def run(self, argv=None, raises=False):
 
         with self._exit_stack:
             try:
-                self.log.info(f"Starting: {self.name}")
+                self.log.info("Starting: %s", self.name)
                 Provenance().start_activity(self.name)
 
                 self.initialize(argv)
 
                 self.setup()
                 self.is_setup = True
 
-                self.log.debug(f"CONFIG: {self.get_current_config()}")
+                self.log.debug("CONFIG: %s", self.get_current_config())
                 Provenance().add_config(self.get_current_config())
 
                 # check for any traitlets warnings using our custom handler
@@ -425,26 +426,44 @@ def run(self, argv=None, raises=False):
 
                 self.start()
                 self.finish()
-                self.log.info(f"Finished: {self.name}")
+                self.log.info("Finished: %s", self.name)
                 Provenance().finish_activity(activity_name=self.name)
             except (ToolConfigurationError, TraitError) as err:
                 self.log.error("%s", err)
                 self.log.error("Use --help for more info")
                 exit_status = 2  # wrong cmd line parameter
+                Provenance().finish_activity(
+                    activity_name=self.name, status="error", exit_code=exit_status
+                )
                 if raises:
                     raise
             except KeyboardInterrupt:
                 self.log.warning("WAS INTERRUPTED BY CTRL-C")
+                exit_status = 130  # Script terminated by Control-C
                 Provenance().finish_activity(
-                    activity_name=self.name, status="interrupted"
+                    activity_name=self.name, status="interrupted", exit_code=exit_status
                 )
-                exit_status = 130  # Script terminated by Control-C
             except Exception as err:
-                self.log.exception(f"Caught unexpected exception: {err}")
-                Provenance().finish_activity(activity_name=self.name, status="error")
+                self.log.exception("Caught unexpected exception: %s", err)
                 exit_status = 1  # any other error
+                Provenance().finish_activity(
+                    activity_name=self.name, status="error", exit_code=exit_status
+                )
                 if raises:
                     raise
+            except SystemExit as err:
+                if raises:
+                    raise  # do not re-intercept in tests
+                else:
+                    exit_status = err.code
+                    self.log.exception(
+                        "Caught SystemExit with exit code %s", exit_status
+                    )
+                    Provenance().finish_activity(
+                        activity_name=self.name,
+                        status="error",
+                        exit_code=exit_status,
+                    )
             finally:
                 if not {"-h", "--help", "--help-all"}.intersection(self.argv):
                     self.write_provenance()